From fec56226a36a04d19716b8a2bb64c20c489ad56e Mon Sep 17 00:00:00 2001 From: Anton Sauchyk Date: Mon, 27 Jul 2026 15:26:03 +0200 Subject: [PATCH 1/2] Send no backlog when the requested sequence number is past its end A client that connects with an Arbitrum-Requested-Sequence-Number after the end of the backlog is already ahead of everything the broadcaster has, so it should be sent nothing. Instead it was sent the entire backlog: backlog.Lookup is an index probe that fails for a number above the tail exactly as it fails for one below the head, and the lookup failure fell back to the backlog head. Bound the requested sequence number against the end of the backlog before the lookup, and record that end as the last sequence number sent so that the catch up after registration does not treat the whole backlog as a gap and resend it through backlog.Get. The end of the backlog is recorded rather than the requested number, which can be arbitrarily far ahead and would then drop every message sent to the client. This restores the behaviour specified in #883, which the old sequencenumbercatchupbuffer implemented explicitly and tested. The guard and its coverage were dropped in the WebSocket library refactor in #1930. Co-Authored-By: Claude Opus 5 (1M context) --- broadcastclient/broadcastclient_test.go | 197 ++++++++++++++++++ broadcaster/backlog/backlog.go | 6 + ...-feed-requested-seqnum-past-backlog-end.md | 2 + wsbroadcastserver/clientconnection.go | 23 +- 4 files changed, 226 insertions(+), 2 deletions(-) create mode 100644 changelog/smypmsa-feed-requested-seqnum-past-backlog-end.md diff --git a/broadcastclient/broadcastclient_test.go b/broadcastclient/broadcastclient_test.go index aec68b3a2a1..5c00a3f0292 100644 --- a/broadcastclient/broadcastclient_test.go +++ b/broadcastclient/broadcastclient_test.go @@ -933,6 +933,203 @@ func connectAndGetCachedMessages(ctx context.Context, addr net.Addr, chainId uin }() } +// awaitSeqNum waits until a message with the given sequence number has been +// received. +func awaitSeqNum(t *testing.T, ts *accumulatingTransactionStreamer, seqNum arbutil.MessageIndex, timeout time.Duration) { + t.Helper() + deadline := time.After(timeout) + for { + for _, msg := range ts.getMessages() { + if msg.SequenceNumber == seqNum { + return + } + } + select { + case <-deadline: + t.Fatalf("timed out waiting for message with sequence number %d", seqNum) + case <-time.After(10 * time.Millisecond): + } + } +} + +// awaitBacklogCount waits until the broadcaster's backlog holds count messages. +func awaitBacklogCount(t *testing.T, b *broadcaster.Broadcaster, count int, timeout time.Duration) { + t.Helper() + deadline := time.After(timeout) + for { + if b.GetCachedMessageCount() == count { + return + } + select { + case <-deadline: + t.Fatalf("timed out waiting for %d messages in the backlog, got %d", count, b.GetCachedMessageCount()) + case <-time.After(10 * time.Millisecond): + } + } +} + +// awaitClientCount waits until count clients have registered with the +// broadcaster. A client only registers once it has been sent the backlog, so +// this also marks the point after which a broadcast reaches the client as a +// live message rather than through the backlog. +func awaitClientCount(t *testing.T, b *broadcaster.Broadcaster, count int32, timeout time.Duration) { + t.Helper() + deadline := time.After(timeout) + for { + if b.ClientCount() == count { + return + } + select { + case <-deadline: + t.Fatalf("timed out waiting for %d clients, got %d", count, b.ClientCount()) + case <-time.After(10 * time.Millisecond): + } + } +} + +// TestBroadcasterRequestedSequenceNumber checks which of the cached messages a +// client is sent for the sequence number it requests. A client that requests a +// sequence number after the end of the backlog is already up to date, so it +// must be sent none of the backlog, and must still be sent the messages that +// follow it. +func TestBroadcasterRequestedSequenceNumber(t *testing.T) { + t.Parallel() + + // The backlog is populated from firstSeqNum rather than 0 so that a + // sequence number before the start of the backlog can be requested. + const firstSeqNum = 10 + const backlogCount = 5 + const lastSeqNum = firstSeqNum + backlogCount - 1 + const sentinelSeqNum = lastSeqNum + 1 + + for _, tc := range []struct { + name string + // requestedSeqNum is sent to the broadcaster in the + // Arbitrum-Requested-Sequence-Number header. A client that has not + // received any messages requests 0, which is what a client that omits + // the header entirely is treated as requesting. + requestedSeqNum uint64 + // expectedFromBacklog are the sequence numbers the client is expected + // to be sent from the backlog, in order. + expectedFromBacklog []uint64 + }{ + { + name: "noneRequestedSendsEntireBacklog", + requestedSeqNum: 0, + expectedFromBacklog: []uint64{10, 11, 12, 13, 14}, + }, + { + name: "beforeBacklogStartSendsEntireBacklog", + requestedSeqNum: firstSeqNum - 5, + expectedFromBacklog: []uint64{10, 11, 12, 13, 14}, + }, + { + name: "withinBacklogSendsFromRequested", + requestedSeqNum: firstSeqNum + 2, + expectedFromBacklog: []uint64{12, 13, 14}, + }, + { + name: "atBacklogEndSendsLastMessage", + requestedSeqNum: lastSeqNum, + expectedFromBacklog: []uint64{14}, + }, + { + name: "afterBacklogEndSendsNothing", + requestedSeqNum: lastSeqNum + 1, + expectedFromBacklog: nil, + }, + { + name: "maxUint64SendsNothing", + requestedSeqNum: ^uint64(0), + expectedFromBacklog: nil, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + chainId := uint64(9743) + + privateKey, err := crypto.GenerateKey() + Require(t, err) + sequencerAddr := crypto.PubkeyToAddress(privateKey.PublicKey) + dataSigner := signature.DataSignerFromPrivateKey(privateKey) + + settings := wsbroadcastserver.DefaultTestBroadcasterConfig + feedErrChan := make(chan error, 10) + b := broadcaster.NewBroadcaster(func() *wsbroadcastserver.BroadcasterConfig { return &settings }, chainId, feedErrChan, dataSigner) + Require(t, b.Initialize()) + Require(t, b.Start(ctx)) + defer b.StopAndWait() + + // Fill the backlog before the client connects, so that the + // requested sequence number is resolved against a backlog of + // firstSeqNum to lastSeqNum. + for seqNum := firstSeqNum; seqNum <= lastSeqNum; seqNum++ { + // #nosec G115 + Require(t, b.BroadcastFeedMessages(feedMessage(t, b, arbutil.MessageIndex(seqNum)))) + } + awaitBacklogCount(t, b, backlogCount, 10*time.Second) + + ts := &accumulatingTransactionStreamer{} + clientFeedErrChan := make(chan error, 10) + broadcastClient, err := newTestBroadcastClient( + DefaultTestConfig, + b.ListenerAddr(), + chainId, + arbutil.MessageIndex(tc.requestedSeqNum), + ts, + nil, + clientFeedErrChan, + &sequencerAddr, + t, + ) + Require(t, err) + broadcastClient.Start(ctx) + defer broadcastClient.StopAndWait() + + awaitClientCount(t, b, 1, 10*time.Second) + + // Sentinel: a live message broadcast after the client has been sent + // the backlog. WebSocket messages are ordered, so once the sentinel + // has arrived every message the client was going to be sent from + // the backlog has already arrived, which makes the exact count + // below deterministic. The sentinel also has to arrive at all: a + // client that is sent none of the backlog must not be left unable + // to receive the messages that follow it. + Require(t, b.BroadcastFeedMessages(feedMessage(t, b, sentinelSeqNum))) + + // Waiting for the sentinel rather than for a message count means the + // client has been sent everything it is going to be sent, whether + // or not that is what is expected. + awaitSeqNum(t, ts, sentinelSeqNum, 10*time.Second) + + expected := append(append([]uint64{}, tc.expectedFromBacklog...), sentinelSeqNum) + var got []uint64 + for _, msg := range ts.getMessages() { + got = append(got, uint64(msg.SequenceNumber)) + } + if len(got) != len(expected) { + t.Fatalf("requested sequence number %d: expected messages %v, got %v", tc.requestedSeqNum, expected, got) + } + for i, seqNum := range expected { + if got[i] != seqNum { + t.Fatalf("requested sequence number %d: expected messages %v, got %v", tc.requestedSeqNum, expected, got) + } + } + + select { + case err := <-clientFeedErrChan: + t.Fatalf("unexpected client feed error: %v", err) + case err := <-feedErrChan: + t.Fatalf("unexpected broadcaster error: %v", err) + default: + } + }) + } +} + func Require(t *testing.T, err error, printables ...interface{}) { t.Helper() testhelpers.RequireImpl(t, err, printables...) diff --git a/broadcaster/backlog/backlog.go b/broadcaster/backlog/backlog.go index 42566ba3efc..cd5dbf913d4 100644 --- a/broadcaster/backlog/backlog.go +++ b/broadcaster/backlog/backlog.go @@ -29,6 +29,7 @@ var ( // Backlog defines the interface for backlog. type Backlog interface { Head() BacklogSegment + Tail() BacklogSegment Append(*message.BroadcastMessage) error Get(uint64, uint64) (*message.BroadcastMessage, error) Count() uint64 @@ -59,6 +60,11 @@ func (b *backlog) Head() BacklogSegment { return b.head.Load() } +// Tail returns the tail backlogSegment within the backlog. +func (b *backlog) Tail() BacklogSegment { + return b.tail.Load() +} + func (b *backlog) backlogSizeInBytes() (uint64, error) { headSeg := b.head.Load() tailSeg := b.tail.Load() diff --git a/changelog/smypmsa-feed-requested-seqnum-past-backlog-end.md b/changelog/smypmsa-feed-requested-seqnum-past-backlog-end.md new file mode 100644 index 00000000000..90fe7c09af6 --- /dev/null +++ b/changelog/smypmsa-feed-requested-seqnum-past-backlog-end.md @@ -0,0 +1,2 @@ +### Fixed +- The feed broadcaster no longer sends the entire backlog to a client that requests a sequence number after the end of the backlog. Such a client is already up to date, so it is now sent none of the backlog and only the messages that follow. diff --git a/wsbroadcastserver/clientconnection.go b/wsbroadcastserver/clientconnection.go index 737d8b61856..3922930f9b7 100644 --- a/wsbroadcastserver/clientconnection.go +++ b/wsbroadcastserver/clientconnection.go @@ -204,8 +204,27 @@ func (cc *ClientConnection) Start(parentCtx context.Context) { // case the backlog is very large segment := cc.backlog.Head() if !backlog.IsBacklogSegmentNil(segment) && segment.Start() < uint64(cc.requestedSeqNum) { - s, err := cc.backlog.Lookup(uint64(cc.requestedSeqNum)) - if err != nil { + // The end of the backlog only has to be read once: a concurrent + // Append can only move it forward and any messages added after this + // read are sent by the catch up below, so the re-read that Get uses + // to avoid racing with a delete is not needed here. + var backlogEnd uint64 + if tail := cc.backlog.Tail(); !backlog.IsBacklogSegmentNil(tail) { + backlogEnd = tail.End() + } + + if uint64(cc.requestedSeqNum) > backlogEnd { + // The client has requested a sequence number after the end of + // the backlog, so there is nothing in the backlog to send. The + // end of the backlog is recorded as the last sequence number + // sent so that the catch up below does not treat the whole + // backlog as a gap and send it anyway. The requested sequence + // number is not used for this as it can be arbitrarily far + // ahead, which would drop every message sent to the client. + log.Debug("client requested sequence number after the end of the backlog, no backlog to send", "client", cc.Name, "requestedSeqNum", cc.requestedSeqNum, "backlogEnd", backlogEnd) + cc.LastSentSeqNum.Store(backlogEnd) + segment = nil + } else if s, err := cc.backlog.Lookup(uint64(cc.requestedSeqNum)); err != nil { logWarn(err, "error finding requested sequence number in backlog: sending the entire backlog instead") } else { segment = s From 52e85794536559c514dd2fcbd09f7da7fb9eefb7 Mon Sep 17 00:00:00 2001 From: Anton Sauchyk Date: Mon, 3 Aug 2026 22:50:51 +0200 Subject: [PATCH 2/2] Treat a zero backlog end as unknown, not as the real end backlog.Append publishes a tail segment before appending to it, so a segment read in that window is empty and End reports zero for it. A client connecting then computed a backlog end of zero, took the past the end branch, was sent none of the backlog, and had zero recorded as the last sequence number sent. The catch up on the next broadcast then treated the whole backlog as a gap and sent it, which is the amplification the past the end check exists to prevent. Fall through to Lookup when the end reads as zero instead, which is what a client connecting in that window was sent before the check existed. Also log the past the end case at warn rather than debug, matching the warning the condition this check replaces used to emit. Co-Authored-By: Claude Opus 5 (1M context) --- wsbroadcastserver/clientconnection.go | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/wsbroadcastserver/clientconnection.go b/wsbroadcastserver/clientconnection.go index 3922930f9b7..a0b9fabdc38 100644 --- a/wsbroadcastserver/clientconnection.go +++ b/wsbroadcastserver/clientconnection.go @@ -204,16 +204,17 @@ func (cc *ClientConnection) Start(parentCtx context.Context) { // case the backlog is very large segment := cc.backlog.Head() if !backlog.IsBacklogSegmentNil(segment) && segment.Start() < uint64(cc.requestedSeqNum) { - // The end of the backlog only has to be read once: a concurrent - // Append can only move it forward and any messages added after this - // read are sent by the catch up below, so the re-read that Get uses - // to avoid racing with a delete is not needed here. + // The end only has to be read once, as messages appended after + // this read are sent by the catch up below. A zero end means + // unknown rather than zero: Append publishes a tail segment before + // appending to it, so a segment read in that window is empty and + // End reports zero for it. Lookup handles a zero end instead. var backlogEnd uint64 if tail := cc.backlog.Tail(); !backlog.IsBacklogSegmentNil(tail) { backlogEnd = tail.End() } - if uint64(cc.requestedSeqNum) > backlogEnd { + if backlogEnd != 0 && uint64(cc.requestedSeqNum) > backlogEnd { // The client has requested a sequence number after the end of // the backlog, so there is nothing in the backlog to send. The // end of the backlog is recorded as the last sequence number @@ -221,7 +222,7 @@ func (cc *ClientConnection) Start(parentCtx context.Context) { // backlog as a gap and send it anyway. The requested sequence // number is not used for this as it can be arbitrarily far // ahead, which would drop every message sent to the client. - log.Debug("client requested sequence number after the end of the backlog, no backlog to send", "client", cc.Name, "requestedSeqNum", cc.requestedSeqNum, "backlogEnd", backlogEnd) + log.Warn("client requested sequence number after the end of the backlog, no backlog to send", "client", cc.Name, "requestedSeqNum", cc.requestedSeqNum, "backlogEnd", backlogEnd) cc.LastSentSeqNum.Store(backlogEnd) segment = nil } else if s, err := cc.backlog.Lookup(uint64(cc.requestedSeqNum)); err != nil {