path: root/core/tx_journal.go

// Copyright 2017 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package core

import (
    "errors"
    "io"
    "os"

    "github.com/ethereum/go-ethereum/common"
    "github.com/ethereum/go-ethereum/core/types"
    "github.com/ethereum/go-ethereum/log"
    "github.com/ethereum/go-ethereum/rlp"
)

// errNoActiveJournal is returned if a transaction is attempted to be inserted
// into the journal, but no such file is currently open.
var errNoActiveJournal = errors.New("no active journal")

// txJournal is a rotating log of transactions with the aim of storing locally
// created transactions to allow non-executed ones to survive node restarts.
type txJournal struct {
    path   string         // Filesystem path to store the transactions at
    writer io.WriteCloser // Output stream to write new transactions into
}

// newTxJournal creates a new transaction journal to
func newTxJournal(path string) *txJournal {
    return &txJournal{
        path: path,
    }
}

// load parses a transaction journal dump from disk, loading its contents into
// the specified pool.
func (journal *txJournal) load(add func(*types.Transaction) error) error {
    // Skip the parsing if the journal file doens't exist at all
    if _, err := os.Stat(journal.path); os.IsNotExist(err) {
        return nil
    }
    // Open the journal for loading any past transactions
    input, err := os.Open(journal.path)
    if err != nil {
        return err
    }
    defer input.Close()

    // Inject all transactions from the journal into the pool
    stream := rlp.NewStream(input, 0)
    total, dropped := 0, 0

    var failure error
    for {
        // Parse the next transaction and terminate on error
        tx := new(types.Transaction)
        if err = stream.Decode(tx); err != nil {
            if err != io.EOF {
                failure = err
            }
            break
        }
        // Import the transaction and bump the appropriate progress counters
        total++
        if err = add(tx); err != nil {
            log.Debug("Failed to add journaled transaction", "err", err)
            dropped++
            continue
        }
    }
    log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)

    return failure
}

// insert adds the specified transaction to the local disk journal.
func (journal *txJournal) insert(tx *types.Transaction) error {
    if journal.writer == nil {
        return errNoActiveJournal
    }
    if err := rlp.Encode(journal.writer, tx); err != nil {
        return err
    }
    return nil
}

// rotate regenerates the transaction journal based on the current contents of
// the transaction pool.
func (journal *txJournal) rotate(all map[common.Address]types.Transactions) error {
    // Close the current journal (if any is open)
    if journal.writer != nil {
        if err := journal.writer.Close(); err != nil {
            return err
        }
        journal.writer = nil
    }
    // Generate a new journal with the contents of the current pool
    replacement, err := os.OpenFile(journal.path+".new", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0755)
    if err != nil {
        return err
    }
    journaled := 0
    for _, txs := range all {
        for _, tx := range txs {
            if err = rlp.Encode(replacement, tx); err != nil {
                replacement.Close()
                return err
            }
        }
        journaled += len(txs)
    }
    replacement.Close()

    // Replace the live journal with the newly generated one
    if err = os.Rename(journal.path+".new", journal.path); err != nil {
        return err
    }
    sink, err := os.OpenFile(journal.path, os.O_WRONLY|os.O_APPEND, 0755)
    if err != nil {
        return err
    }
    journal.writer = sink
    log.Info("Regenerated local transaction journal", "transactions", journaled, "accounts", len(all))

    return nil
}

// close flushes the transaction journal contents to disk and closes the file.
func (journal *txJournal) close() error {
    var err error

    if journal.writer != nil {
        err = journal.writer.Close()
        journal.writer = nil
    }
    return err
}