Merge pull request #3793 from ethereum-optimism/jg/frame_validity_reqs

op-node: Add frame validity checks 

op-node/rollup/derive/channel.go

@@ -10,9 +10,6 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
-// TODO: Full state machine for channel
-// Open, Closed, Ready (todo - when to construct RLP reader)
-
 // A Channel is a set of batches that are split into at least one, but possibly multiple frames.
 // Frames are allowed to be ingested out of order.
 // Each frame is ingested one by one. Once a frame with `closed` is added to the channel, the
@@ -28,17 +25,15 @@ type Channel struct {
 	// true if we have buffered the last frame
 	closed bool
 
-	// TODO: implement this check
 	// highestFrameNumber is the highest frame number yet seen.
-	// This must be equal to `endFrameNumber`
-	// highestFrameNumber uint16
+	highestFrameNumber uint16
 
 	// endFrameNumber is the frame number of the frame where `isLast` is true
 	// No other frame number must be larger than this.
 	endFrameNumber uint16
 
-	// Store a map of frame number -> frame data for constant time ordering
-	inputs map[uint64][]byte
+	// Store a map of frame number -> frame for constant time ordering
+	inputs map[uint64]Frame
 
 	highestL1InclusionBlock eth.L1BlockRef
 }
@@ -46,7 +41,7 @@ type Channel struct {
 func NewChannel(id ChannelID, openBlock eth.L1BlockRef) *Channel {
 	return &Channel{
 		id:        id,
-		inputs:    make(map[uint64][]byte),
+		inputs:    make(map[uint64]Frame),
 		openBlock: openBlock,
 	}
 }
@@ -58,26 +53,43 @@ func (ch *Channel) AddFrame(frame Frame, l1InclusionBlock eth.L1BlockRef) error
 	if frame.ID != ch.id {
 		return fmt.Errorf("frame id does not match channel id. Expected %v, got %v", ch.id, frame.ID)
 	}
+	// These checks are specified and cannot be changed without a hard fork.
 	if frame.IsLast && ch.closed {
 		return fmt.Errorf("cannot add ending frame to a closed channel. id %v", ch.id)
 	}
 	if _, ok := ch.inputs[uint64(frame.FrameNumber)]; ok {
 		return DuplicateErr
 	}
-	// TODO: highest seen frame vs endFrameNumber
+	if ch.closed && frame.FrameNumber >= ch.endFrameNumber {
+		return fmt.Errorf("frame number (%d) is greater than or equal to end frame number (%d) of a closed channel", frame.FrameNumber, ch.endFrameNumber)
+	}
 
 	// Guaranteed to succeed. Now update internal state
 	if frame.IsLast {
 		ch.endFrameNumber = frame.FrameNumber
 		ch.closed = true
 	}
+	// Prune frames with a number higher than the closing frame number when we receive a closing frame
+	if frame.IsLast && ch.endFrameNumber < ch.highestFrameNumber {
+		// Do a linear scan over saved inputs instead of ranging over ID numbers
+		for id, prunedFrame := range ch.inputs {
+			if id >= uint64(ch.endFrameNumber) {
+				delete(ch.inputs, id)
+			}
+			ch.size -= frameSize(prunedFrame)
+		}
+		ch.highestFrameNumber = ch.endFrameNumber
+	}
+	// Update highest seen frame number after pruning
+	if frame.FrameNumber > ch.highestFrameNumber {
+		ch.highestFrameNumber = frame.FrameNumber
+	}
+
 	if ch.highestL1InclusionBlock.Number < l1InclusionBlock.Number {
 		ch.highestL1InclusionBlock = l1InclusionBlock
 	}
-	ch.inputs[uint64(frame.FrameNumber)] = frame.Data
-	ch.size += uint64(len(frame.Data)) + frameOverhead
-
-	// todo use `IsReady` + state to create final output reader
+	ch.inputs[uint64(frame.FrameNumber)] = frame
+	ch.size += frameSize(frame)
 
 	return nil
 }
@@ -121,11 +133,11 @@ func (ch *Channel) IsReady() bool {
 func (ch *Channel) Reader() io.Reader {
 	var readers []io.Reader
 	for i := uint64(0); i <= uint64(ch.endFrameNumber); i++ {
-		data, ok := ch.inputs[i]
+		frame, ok := ch.inputs[i]
 		if !ok {
 			panic("dev error in channel.Reader. Must be called after the channel is ready.")
 		}
-		readers = append(readers, bytes.NewBuffer(data))
+		readers = append(readers, bytes.NewReader(frame.Data))
 	}
 	return io.MultiReader(readers...)
 }

op-node/rollup/derive/channel_test.go

+package derive
+
+import (
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/stretchr/testify/require"
+)
+
+type frameValidityTC struct {
+	name      string
+	frames    []Frame
+	shouldErr []bool
+	sizes     []uint64
+}
+
+func (tc *frameValidityTC) Run(t *testing.T) {
+	id := [16]byte{0xff}
+	block := eth.L1BlockRef{}
+	ch := NewChannel(id, block)
+
+	if len(tc.frames) != len(tc.shouldErr) || len(tc.frames) != len(tc.sizes) {
+		t.Errorf("lengths should be the same. frames: %d, shouldErr: %d, sizes: %d", len(tc.frames), len(tc.shouldErr), len(tc.sizes))
+	}
+
+	for i, frame := range tc.frames {
+		err := ch.AddFrame(frame, block)
+		if tc.shouldErr[i] {
+			require.NotNil(t, err)
+		} else {
+			require.Nil(t, err)
+		}
+		require.Equal(t, tc.sizes[i], ch.Size())
+	}
+}
+
+// TestFrameValidity inserts a list of frames into the channel. It checks if an error
+// should be returned by `AddFrame` as well as checking the size of the channel.
+func TestFrameValidity(t *testing.T) {
+	id := [16]byte{0xff}
+	testCases := []frameValidityTC{
+		{
+			name:      "wrong channel",
+			frames:    []Frame{{ID: [16]byte{0xee}}},
+			shouldErr: []bool{true},
+			sizes:     []uint64{0},
+		},
+		{
+			name: "double close",
+			frames: []Frame{
+				{ID: id, FrameNumber: 2, IsLast: true, Data: []byte("four")},
+				{ID: id, FrameNumber: 1, IsLast: true}},
+			shouldErr: []bool{false, true},
+			sizes:     []uint64{204, 204},
+		},
+		{
+			name: "duplicate frame",
+			frames: []Frame{
+				{ID: id, FrameNumber: 2, Data: []byte("four")},
+				{ID: id, FrameNumber: 2, Data: []byte("seven__")}},
+			shouldErr: []bool{false, true},
+			sizes:     []uint64{204, 204},
+		},
+		{
+			name: "duplicate closing frames",
+			frames: []Frame{
+				{ID: id, FrameNumber: 2, IsLast: true, Data: []byte("four")},
+				{ID: id, FrameNumber: 2, IsLast: true, Data: []byte("seven__")}},
+			shouldErr: []bool{false, true},
+			sizes:     []uint64{204, 204},
+		},
+		{
+			name: "frame past closing",
+			frames: []Frame{
+				{ID: id, FrameNumber: 2, IsLast: true, Data: []byte("four")},
+				{ID: id, FrameNumber: 10, Data: []byte("seven__")}},
+			shouldErr: []bool{false, true},
+			sizes:     []uint64{204, 204},
+		},
+		{
+			name: "prune after close frame",
+			frames: []Frame{
+				{ID: id, FrameNumber: 10, IsLast: false, Data: []byte("seven__")},
+				{ID: id, FrameNumber: 2, IsLast: true, Data: []byte("four")}},
+			shouldErr: []bool{false, false},
+			sizes:     []uint64{207, 204},
+		},
+		{
+			name: "multiple valid frames",
+			frames: []Frame{
+				{ID: id, FrameNumber: 10, Data: []byte("seven__")},
+				{ID: id, FrameNumber: 2, Data: []byte("four")}},
+			shouldErr: []bool{false, false},
+			sizes:     []uint64{207, 411},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, tc.Run)
+	}
+}

op-node/rollup/derive/params.go

@@ -8,6 +8,14 @@ import (
 // count the tagging info as 200 in terms of buffer size.
 const frameOverhead = 200
 
+// frameSize calculates the size of the frame + overhead for
+// storing the frame. The sum of the frame size of each frame in
+// a channel determines the channel's size. The sum of the channel
+// sizes is used for pruning & compared against `MaxChannelBankSize`
+func frameSize(frame Frame) uint64 {
+	return uint64(len(frame.Data)) + frameOverhead
+}
+
 const DerivationVersion0 = 0
 
 // MaxChannelBankSize is the amount of memory space, in number of bytes,

specs/derivation.md

@@ -464,18 +464,18 @@ particular we extract a byte string that corresponds to the concatenation of the
 transaction][g-batcher-transaction] belonging to the block. This byte stream encodes a stream of [channel
 frames][g-channel-frame] (see the [Batch Submission Wire Format][wire-format] section for more info).
 
-These frames are parsed, then grouped per [channel][g-channel] into a structure we call the *channel bank*.
+These frames are parsed, then grouped per [channel][g-channel] into a structure we call the *channel bank*. When
+adding frames the the channel, individual frames may be invalid, but the channel does not have a notion of validity
+until the channel timeout is up. This enables adding the option to do a partial read from the channel in the future.
 
 Some frames are ignored:
 
-- Frames where `frame.frame_number <= highest_frame_number`, where `highest_frame_number` is the highest frame number
-  that was previously encountered for this channel.
-  - i.e. in case of duplicate frame, the first frame read from L1 is considered canonical.
-- Frames with a higher number than that of the final frame of the channel (i.e. the first frame marked with
-  `frame.is_last == 1`) are ignored.
-  - These frames could still be written into the channel bank if we haven't seen the final frame yet. But they will
-      never be read out from the channel bank.
-- Frames with a channel ID whose timestamp are higher than that of the L1 block on which the frame appears.
+- Frames with the same frame number as an existing frame in the channel (a duplicate). The first seen frame is used.
+- Frames that attempt to close an already closed channel. This would be the second frame with `frame.is_last == 1` even
+  if the frame number of the second frame is not the same as the first frame which closed the channel.
+
+If a frame with `is_last == 1` is added to a channel, all frames with a higher frame number are removed from the
+channel.
 
 Channels are also recorded in FIFO order in a structure called the *channel queue*. A channel is added to the channel
 queue the first time a frame belonging to the channel is seen. This structure is used in the next stage.