eth/sequencer: cap coalesced published records at the store message limit (#2400)

The send path coalesces up to 64 transactions into one published record
(commit 89143d7). It bounded a record only by that count and the 32 MB
pending-input limit, never by the store's Redpanda max.message.bytes, so a
run of large transactions (e.g. 64 x 32 KiB) produced a >1 MiB record. That
draws a permanent MESSAGE_TOO_LARGE on produce; the ingress treats a produce
failure as a fence and takeover replays the same record, so one oversized
record wedges the writer and nothing behind it is preconfirmed.

Cap a coalesced record at maxRecordBytes (max.message.bytes less a 4 KiB
reserve for per-transaction protobuf framing, the prefix commitment, and the
ingress's own recordFraming allowance), so bor never builds a record the
ingress would reject. A lone transaction over the cap is still emitted and
left for the store to judge, not dropped.

Also give the terminal MALFORMED ack path a clearer message: with the cap in
place a MALFORMED means the store's max.message.bytes is below this build's
record cap, not an oversized record.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: cffls

eth/sequencer/stream.go

@@ -63,6 +63,17 @@ type sent struct {
 
 const maxTransactionsPerPublishedRecord = 64
 
+// maxMessageBytes mirrors the store's Redpanda max.message.bytes: a larger
+// record draws a permanent MESSAGE_TOO_LARGE on produce and fails the publisher.
+const maxMessageBytes = 1 << 20
+
+// recordSizeReserve leaves headroom for per-transaction protobuf framing and the
+// prefix commitment, so a capped record's proto.Size stays under what the ingress
+// accepts (max.message.bytes less its own recordFraming).
+const recordSizeReserve = 4 * 1024
+
+const maxRecordBytes = maxMessageBytes - recordSizeReserve
+
 const publishedRecordBatchDelay = 5 * time.Millisecond
 
 // stallTracker watches ack progress from outside the send loop: a hung
@@ -344,7 +355,9 @@ func coalescePublishedRecord(items []journalItem, acked uint64) (*pb.Entry, []jo
 		for _, raw := range candidate.GetTransactions() {
 			candidateBytes += uint64(len(raw))
 		}
-		if candidateBytes > pendingInputLimit-inputBytes {
+		// A lone transaction over the cap still ships via the single-record
+		// return below; the store judges it rather than losing it here.
+		if inputBytes+candidateBytes > uint64(maxRecordBytes) {
 			break
 		}
 		batch = append(batch, item)
@@ -406,7 +419,9 @@ func (p *Publisher) handleAck(ack ackResult, inflight *[]sent) (streamResult, bo
 		// resending in order is exact.
 		return streamResult{reason: endTransport}, true
 	default:
-		p.fail("store rejected entry", "status", ack.status)
+		// Terminal but safe: no fence, no resend. A MALFORMED here means the
+		// store's max.message.bytes is below our record cap.
+		p.fail("store rejected entry", "status", ack.status, "seq", first.item.seq)
 
 		return streamResult{reason: endTerminal}, true
 	}

eth/sequencer/stream_test.go

@@ -5,6 +5,8 @@ import (
 	"testing"
 	"time"
 
+	"google.golang.org/protobuf/proto"
+
 	"github.com/0xPolygon/sequence-store-proto/commitment"
 	pb "github.com/0xPolygon/sequence-store-proto/sequencestore/v1"
 )
@@ -182,3 +184,103 @@ func TestHandleAckRetiresPublishedRecordBatch(t *testing.T) {
 		t.Fatalf("acked = %d, anchor = %x", p.ackedSeq, p.anchor)
 	}
 }
+
+func TestCoalesceRecordByteCap(t *testing.T) {
+	// Coalescing many large txs must stay under the store's max.message.bytes.
+	const txSize = 32 * 1024
+
+	items := make([]journalItem, 40)
+	start := commitment.Head{0x01}
+	head := start
+	for i := range items {
+		raw := make([]byte, txSize)
+		raw[0] = byte(i + 1)
+		next := commitment.FoldTx(head, raw)
+		items[i] = journalItem{
+			seq:    uint64(i + 1),
+			entry:  recordEntry(raw, head),
+			pre:    head,
+			post:   next,
+			kind:   entryRecord,
+			height: 9,
+		}
+		head = next
+	}
+
+	entry, batch := coalescePublishedRecord(items, 0)
+
+	if len(batch) == 0 || len(batch) >= len(items) {
+		t.Fatalf("batch length = %d, want a byte-capped prefix of %d", len(batch), len(items))
+	}
+
+	got := 0
+	for _, raw := range entry.GetRecord().GetTransactions() {
+		got += len(raw)
+	}
+	if got > maxRecordBytes {
+		t.Fatalf("coalesced tx bytes = %d, over the %d cap", got, maxRecordBytes)
+	}
+	if got+txSize <= maxRecordBytes {
+		t.Fatalf("batch not maximal: %d plus one more tx still fits under %d", got, maxRecordBytes)
+	}
+
+	// proto.Size must stay under what the ingress accepts (max.message.bytes
+	// less its recordFraming allowance).
+	if size := proto.Size(entry); size > maxMessageBytes-1024 {
+		t.Fatalf("record proto size = %d, the ingress rejects above %d", size, maxMessageBytes-1024)
+	}
+
+	if fold := commitment.FoldTxs(start, entry.GetRecord().GetTransactions()); fold != items[len(batch)-1].post {
+		t.Fatalf("coalesced commitment = %x, want %x", fold, items[len(batch)-1].post)
+	}
+}
+
+func TestCoalesceRecordCountCap(t *testing.T) {
+	// Small records cap at maxTransactionsPerPublishedRecord, well below the
+	// byte cap.
+	items := make([]journalItem, maxTransactionsPerPublishedRecord+20)
+	head := commitment.Head{0x01}
+	for i := range items {
+		raw := []byte{byte(i%251 + 1)}
+		next := commitment.FoldTx(head, raw)
+		items[i] = journalItem{
+			seq:    uint64(i + 1),
+			entry:  recordEntry(raw, head),
+			pre:    head,
+			post:   next,
+			kind:   entryRecord,
+			height: 9,
+		}
+		head = next
+	}
+
+	if _, batch := coalescePublishedRecord(items, 0); len(batch) != maxTransactionsPerPublishedRecord {
+		t.Fatalf("batch length = %d, want the %d count cap", len(batch), maxTransactionsPerPublishedRecord)
+	}
+}
+
+func TestCoalesceRecordByteCapBoundary(t *testing.T) {
+	// Two records whose sizes sum to exactly maxRecordBytes coalesce together:
+	// the cap boundary is inclusive (>, not >=).
+	head := commitment.Head{0x02}
+	sizes := []int{100_000, maxRecordBytes - 100_000}
+	items := make([]journalItem, len(sizes))
+	for i, sz := range sizes {
+		raw := make([]byte, sz)
+		raw[0] = byte(i + 1)
+		next := commitment.FoldTx(head, raw)
+		items[i] = journalItem{
+			seq:    uint64(i + 1),
+			entry:  recordEntry(raw, head),
+			pre:    head,
+			post:   next,
+			kind:   entryRecord,
+			height: 3,
+		}
+		head = next
+	}
+
+	if _, batch := coalescePublishedRecord(items, 0); len(batch) != 2 {
+		t.Fatalf("exact-fill batch = %d, want 2 (the cap boundary is inclusive)", len(batch))
+	}
+}