WIP: Reassemble Channels

This is WIP because it doesn't perfectly re-assemble channels, but starts
to provide info on dangling channels.

op-node/cmd/batch_decoder/README.md

+# Batch Decoding Tool
+
+The batch decoding tool is a utility to aid in debugging the batch submitter & the op-node
+by looking at what batches were submitted on L1.
+
+## Design Philosophy
+
+The `batch_decoder` tool is designed to be simple & flexible. It offloads as much data analysis
+as possible to other tools. It is built around manipulating JSON on disk. The first stage is to
+fetch all transaction which are sent to a batch inbox address. Those transactions are decoded into
+frames in that step & information about them is recorded. After transactions are fetched the frames
+are re-assembled into channels in a second step that does not touch the network.
+
+
+## Commands
+
+### Fetch
+
+`batch_decoder fetch` pulls all L1 transactions sent to the batch inbox address in a given L1 block
+range and then stores them on disk to a specified path as JSON files where the name of the file is
+the transaction hash.
+
+### Reassemble
+
+`batch_decoder reassemble` goes through all of the found frames in the cache & then turns them
+into channels. It then stores the channels with metadata on disk where the file name is the Channel ID.
+
+
+## JQ Cheat Sheet
+
+`jq` is a really useful utility for manipulating JSON files.
+
+```
+# Pretty print a JSON file
+jq . $JSON_FILE
+
+# Print the number of valid & invalid transactions
+jq .valid_data $TX_DIR/* | sort | uniq -c  
+
+# Select all transactions that have invalid data & then print the transaction hash
+jq "select(.valid_data == false)|.tx.hash" $TX_DIR
+
+# Select all channels that are not ready and then get the id and inclusion block & tx hash of the first frame.
+jq "select(.is_ready == false)|[.id, .frames[0].inclusion_block, .frames[0].transaction_hash]"  $CHANNEL_DIR
+```
+
+
+## Roadmap
+
+- Parallel transaction fetching (CLI-3563)
+- Create force-close channel tx data from channel ID (CLI-3564)
+- Better re-assembly of channels (CLI-3559)
+- Pull the batches out of channels & store that information inside the ChannelWithMetadata (CLI-3565)
+	- Transaction Bytes used
+	- Total uncompressed (different from tx bytes) + compressed bytes
+- Invert ChannelWithMetadata so block numbers/hashes are mapped to channels they are submitted in (CLI-3560)

op-node/cmd/batch_decoder/fetch/fetch.go

@@ -38,6 +38,9 @@ type Config struct {
 	OutDirectory string
 }
 
+// Batches fetches & stores all transactions sent to the batch inbox address in
+// the given block range (inclusive to exclusive).
+// The transactions & metadata are written to the out directory.
 func Batches(client *ethclient.Client, config Config) (totalValid, totalInvalid int) {
 	if err := os.MkdirAll(config.OutDirectory, 0750); err != nil {
 		log.Fatal(err)
@@ -53,13 +56,15 @@ func Batches(client *ethclient.Client, config Config) (totalValid, totalInvalid
 	return
 }
 
+// fetchBatchesPerBlock gets a block & the parses all of the transactions in the block.
 func fetchBatchesPerBlock(client *ethclient.Client, number *big.Int, signer types.Signer, config Config) (validBatchCount, invalidBatchCount int) {
-	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 	defer cancel()
 	block, err := client.BlockByNumber(ctx, number)
 	if err != nil {
 		log.Fatal(err)
 	}
+	fmt.Println("Fetched block: ", number)
 	for i, tx := range block.Transactions() {
 		if tx.To() != nil && *tx.To() == config.BatchInbox {
 			sender, err := signer.Sender(tx)

op-node/cmd/batch_decoder/main.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/ethereum-optimism/optimism/op-node/cmd/batch_decoder/fetch"
+	"github.com/ethereum-optimism/optimism/op-node/cmd/batch_decoder/reassemble"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/urfave/cli"
@@ -59,7 +60,7 @@ func main() {
 				if err != nil {
 					log.Fatal(err)
 				}
-				ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+				ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 				defer cancel()
 				chainID, err := client.ChainID(ctx)
 				if err != nil {
@@ -82,6 +83,36 @@ func main() {
 				return nil
 			},
 		},
+		{
+			Name:  "reassemble",
+			Usage: "Reassembles channels from fetched batches",
+			Flags: []cli.Flag{
+				cli.StringFlag{
+					Name:  "inbox",
+					Value: "0xff00000000000000000000000000000000000420",
+					Usage: "Batch Inbox Address",
+				},
+				cli.StringFlag{
+					Name:  "in",
+					Value: "/tmp/batch_decoder/transactions_cache",
+					Usage: "Cache directory for the found transactions",
+				},
+				cli.StringFlag{
+					Name:  "out",
+					Value: "/tmp/batch_decoder/channel_cache",
+					Usage: "Cache directory for the found channels",
+				},
+			},
+			Action: func(cliCtx *cli.Context) error {
+				config := reassemble.Config{
+					BatchInbox:   common.HexToAddress(cliCtx.String("inbox")),
+					InDirectory:  cliCtx.String("in"),
+					OutDirectory: cliCtx.String("out"),
+				}
+				reassemble.Channels(config)
+				return nil
+			},
+		},
 	}
 
 	if err := app.Run(os.Args); err != nil {

op-node/cmd/batch_decoder/reassemble/reassemble.go

+package reassemble
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"path"
+	"sort"
+
+	"github.com/ethereum-optimism/optimism/op-node/cmd/batch_decoder/fetch"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum/go-ethereum/common"
+)
+
+type ChannelWithMeta struct {
+	ID            derive.ChannelID    `json:"id"`
+	SkippedFrames []FrameWithMetadata `json:"skipped_frames"`
+	IsReady       bool                `json:"is_ready"`
+	Frames        []FrameWithMetadata `json:"frames"`
+}
+
+type FrameWithMetadata struct {
+	TxHash         common.Hash  `json:"transaction_hash"`
+	InclusionBlock uint64       `json:"inclusion_block"`
+	Frame          derive.Frame `json:"frame"`
+}
+
+type Config struct {
+	BatchInbox   common.Address
+	InDirectory  string
+	OutDirectory string
+}
+
+// Channels loads all transactions from the given input directory that are submitted to the
+// specified batch inbox and then re-assembles all channels & writes the re-assembled channels
+// to the out directory.
+func Channels(config Config) {
+	if err := os.MkdirAll(config.OutDirectory, 0750); err != nil {
+		log.Fatal(err)
+	}
+	txns := loadTransactions(config.InDirectory, config.BatchInbox)
+	// Sort first by block number then by transaction index inside the block number range.
+	// This is to match the order they are processed in derivation.
+	sort.Slice(txns, func(i, j int) bool {
+		if txns[i].BlockNumber == txns[j].BlockNumber {
+			return txns[i].TxIndex < txns[j].TxIndex
+		} else {
+			return txns[i].BlockNumber < txns[j].BlockNumber
+		}
+
+	})
+	frames := transactionsToFrames(txns)
+	framesByChannel := make(map[derive.ChannelID][]FrameWithMetadata)
+	for _, frame := range frames {
+		framesByChannel[frame.Frame.ID] = append(framesByChannel[frame.Frame.ID], frame)
+	}
+	for id, frames := range framesByChannel {
+		ch := processFrames(id, frames)
+		filename := path.Join(config.OutDirectory, fmt.Sprintf("%s.json", id.String()))
+		file, err := os.Create(filename)
+		if err != nil {
+			log.Fatal(err)
+		}
+		defer file.Close()
+		enc := json.NewEncoder(file)
+		if err := enc.Encode(ch); err != nil {
+			log.Fatal(err)
+		}
+	}
+}
+
+func processFrames(id derive.ChannelID, frames []FrameWithMetadata) ChannelWithMeta {
+	// TO DO: Use the same approach as in derivation.
+	ready := false
+	for _, frame := range frames {
+		ready = frame.Frame.IsLast || ready
+	}
+	if !ready {
+		fmt.Printf("Found channel that was not closed: %v\n", id.String())
+	}
+	return ChannelWithMeta{
+		ID:            id,
+		Frames:        frames,
+		SkippedFrames: nil,
+		IsReady:       ready,
+	}
+}
+
+func transactionsToFrames(txns []fetch.TransactionWithMeta) []FrameWithMetadata {
+	var out []FrameWithMetadata
+	for _, tx := range txns {
+		for _, frame := range tx.Frames {
+			fm := FrameWithMetadata{
+				TxHash:         tx.Tx.Hash(),
+				InclusionBlock: tx.BlockNumber,
+				Frame:          frame,
+			}
+			out = append(out, fm)
+		}
+	}
+	return out
+}
+
+func loadTransactions(dir string, inbox common.Address) []fetch.TransactionWithMeta {
+	files, err := os.ReadDir(dir)
+	if err != nil {
+		log.Fatal(err)
+	}
+	var out []fetch.TransactionWithMeta
+	for _, file := range files {
+		f, err := os.Open(path.Join(dir, file.Name()))
+		if err != nil {
+			log.Fatal(err)
+		}
+		defer f.Close()
+		dec := json.NewDecoder(f)
+		var txm fetch.TransactionWithMeta
+		if err := dec.Decode(&txm); err != nil {
+			log.Fatalf("Failed to decode %v. Err: %v\n", path.Join(dir, file.Name()), err)
+		}
+		if txm.InboxAddr == inbox && txm.ValidSender {
+			out = append(out, txm)
+		}
+	}
+	return out
+}