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..a0b9fabdc38 100644 --- a/wsbroadcastserver/clientconnection.go +++ b/wsbroadcastserver/clientconnection.go @@ -204,8 +204,28 @@ 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 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 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 + // 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.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 { logWarn(err, "error finding requested sequence number in backlog: sending the entire backlog instead") } else { segment = s