From 93cabeb73e88e5bb91bef799c591181f68d0d4a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Nov 2026 12:34:04 +0000 Subject: [PATCH] fix(Stream): Resume an unfinished replay, ask again for an empty one and retry refused follow-ups - A drop while a replay is catching up resumes from the bar it reached, not from the newest live row, which arrives before `replayed`. - A truncated replay with no bars (`to` null) is asked for again from `from`; empty a second time from the same bar, that bar is larger than the server's cap, and stderr says so instead of "more to come". - A follow-up refused with limit_exceeded is sent again after a jittered pause, up to the busy retries, then the rest is left to REST with a warning. Running out of follow-ups now warns too. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MoDcuxddwYnzAPetT7cR5r --- README.md | 2 +- cli_stream.go | 2 +- stream.go | 135 +++++++++++++++++++++++++++++++++++++----- stream_replay_test.go | 106 +++++++++++++++++++++++++++++++++ 4 files changed, 229 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 3d8b02d..c03ee29 100644 --- a/README.md +++ b/README.md @@ -290,7 +290,7 @@ aperiodic stream ohlcv \ - `data` is the server's row object, unchanged apart from whitespace (no re-escaping). On plans that include it, the latest row per symbol arrives first with `"snapshot":true`. - Granted and rejected channels, server warnings and reconnects go to **stderr**. - Dropped connections reconnect with capped exponential backoff and re-subscribe. This covers network drops, server restarts or drains (closes such as 1000, 1001, 1011, 1012), handshake 5xx and 75 s of silence. The backoff starts over only once a connection has delivered a row or stayed up for 60 s. -- **Replay.** On 1m and longer intervals, a reconnect asks for the bars since the newest one received, so rows published while disconnected arrive first, marked `"replay":true` (the server keeps 1 h of 1m to 30m bars, 3 h of 1h, the newest 4h bar and 48 h of 1d). `--since` replays from a time of your choosing on the first connect, too, and `--replay=false` turns the reconnect replay off. A summary of each replay goes to stderr, including bars the server does not hold, which you fill from the REST API. A row can arrive both replayed and live with the same key: overwrite it. If the server is too busy replaying (after it restarts), the subscribe is sent again after a short random pause; a replay cut short at the server's size cap is followed up from the next bar. Below 1m nothing is replayed. +- **Replay.** On 1m and longer intervals, a reconnect asks for the bars since the newest one received, so rows published while disconnected arrive first, marked `"replay":true` (the server keeps 1 h of 1m to 30m bars, 3 h of 1h, the newest 4h bar and 48 h of 1d). `--since` replays from a time of your choosing on the first connect, too, and `--replay=false` turns the reconnect replay off. A summary of each replay goes to stderr, including bars the server does not hold, which you fill from the REST API. A row can arrive both replayed and live with the same key: overwrite it. If the server is too busy replaying (after it restarts), the subscribe is sent again after a short random pause; a replay cut short at the server's size cap is followed up from the next bar, a follow-up refused while the server is busy is sent again, and a drop while a replay is catching up resumes it from where it got to. Below 1m nothing is replayed. - It exits 1 with the server's message, without retrying, if: - the key is refused (401), the plan has no live access (403), or the handshake gets 426; - the connection limit is reached (429) on the first connect. After a session has been subscribed, a 429 is retried, because the server may still be counting the dropped socket; diff --git a/cli_stream.go b/cli_stream.go index f7e1bcc..c548cb8 100644 --- a/cli_stream.go +++ b/cli_stream.go @@ -165,7 +165,7 @@ func (c *CLI) runStream(args []string) int { if len(r.Missing) > 0 { fmt.Fprintf(c.Stderr, "; %d bars the server does not hold", len(r.Missing)) } - if r.Truncated { + if r.FollowedUp { fmt.Fprint(c.Stderr, "; more to come") } fmt.Fprintln(c.Stderr) diff --git a/stream.go b/stream.go index 6dc87a0..c9e8b00 100644 --- a/stream.go +++ b/stream.go @@ -25,8 +25,10 @@ const ( // for longer than this is treated as dropped. streamIdleTimeout = 75 * time.Second streamDialTimeout = 15 * time.Second - // A replay frame carries one whole bar: on an all-symbols channel that is - // over a megabyte. The server caps one subscribe's replay at a few MB. + // The server splits a replayed bar into small frames, but one row can be + // larger, and earlier servers sent a whole bar (over a megabyte on an + // all-symbols channel) in one frame. The server caps one subscribe's + // replay at a few MB. streamMessageLimit = 16 << 20 // How often a subscribe refused because the server is busy replaying is @@ -93,6 +95,9 @@ type StreamReplay struct { Missing []int64 Truncated bool Unavailable bool + // FollowedUp is true when the stream asks for the rest of a truncated + // replay; a truncated one without it stopped, and the rest is for REST. + FollowedUp bool } type streamFrame struct { @@ -106,6 +111,7 @@ type streamFrame struct { Channel string `json:"channel"` Snapshot bool `json:"snapshot"` Replay bool `json:"replay"` + Time int64 `json:"time"` Data json.RawMessage `json:"data"` From int64 `json:"from"` @@ -190,6 +196,11 @@ type Streamer struct { // replay starts. lastBar map[string]int64 subscribedAt time.Time + // replayFrom is each channel whose replay is not finished, and the bar + // to ask from if it must be asked for again. A drop resumes from here, + // not from lastBar: held live rows come before the replay's end, so + // lastBar runs ahead of a replay still catching up. + replayFrom map[string]int64 } // NewStreamer reads APERIODIC_STREAM_URL and the Cloudflare Access service @@ -225,6 +236,8 @@ func (s *Streamer) subscribeChannels(resume bool) []StreamChannel { switch { case !resume && !s.Since.IsZero(): channels[i].Since = s.Since.UnixMilli() + case resume && s.Replay && s.replayFrom[ch.name()] > 0: + channels[i].Since = s.replayFrom[ch.name()] case resume && s.Replay && s.lastBar[ch.name()] > 0: channels[i].Since = s.lastBar[ch.name()] case resume && s.Replay && !s.subscribedAt.IsZero(): @@ -236,6 +249,49 @@ func (s *Streamer) subscribeChannels(resume bool) []StreamChannel { return channels } +// expectReplays records where each channel subscribed with a since replays +// from, until its replay is finished. +func (s *Streamer) expectReplays(channels []StreamChannel) { + if s.replayFrom == nil { + s.replayFrom = map[string]int64{} + } + for _, ch := range channels { + if ch.Since > 0 { + s.replayFrom[ch.name()] = ch.Since + } + } +} + +// nextReplay decides whether a replayed channel needs a follow-up, and from +// which bar. A truncated replay goes on from the bar after its last; one +// left with no bars by the size cap a subscribe's channels share (To nil) +// goes on from From; empty again from the same bar, alone, that bar is +// larger than the cap, and why says so. +func (s *Streamer) nextReplay(frame streamFrame, stalled map[string]int64, continuations int) (since int64, follow bool, why string) { + step, replayable := streamReplaySteps[channelInterval(frame.Channel)] + if !frame.Truncated || !replayable { + return 0, false, "" + } + if _, tracked := s.replayFrom[frame.Channel]; !tracked { + return 0, false, "" + } + switch { + case frame.To != nil: + since = *frame.To + step + delete(stalled, frame.Channel) + case stalled[frame.Channel] != frame.From: + since = frame.From + stalled[frame.Channel] = frame.From + default: + return 0, false, fmt.Sprintf("a bar of %s is larger than the server's replay cap; fetch it and what follows from the REST API", frame.Channel) + } + if continuations >= streamMaxContinuation { + return 0, false, fmt.Sprintf("the replay of %s is still cut short after %d follow-ups; fill it from %s from the REST API", + frame.Channel, streamMaxContinuation, time.UnixMilli(since).UTC().Format(time.RFC3339)) + } + return since, true, "" +} + // noteBar records a row's bar time, so a reconnect replays from it. func (s *Streamer) noteBar(channel string, data json.RawMessage) { var row struct { @@ -388,9 +444,28 @@ func (s *Streamer) session(ctx context.Context, resume bool) (res sessionResult) return res } s.subscribedAt = time.Now() + s.expectReplays(channels) granted := map[string]bool{} - busyRetries, continuations := 0, 0 + busyRetries, continuations, refusals := 0, 0, 0 + // followUps maps a follow-up's id to its channel; stalled is the bar a + // replay came back empty from (the size cap a subscribe's channels share + // can leave a later one with nothing). + followUps := map[string]string{} + stalled := map[string]int64{} + followUp := func(channel string, since int64) error { + ch, ok := s.channelNamed(channel) + if !ok { + delete(s.replayFrom, channel) + return nil + } + continuations++ + id := fmt.Sprintf("r%d", continuations) + ch.Since = since + followUps[id] = channel + s.replayFrom[channel] = since + return write(id, []StreamChannel{ch}) + } for { msg, err := readStreamMessage(conn) if err != nil { @@ -423,6 +498,9 @@ func (s *Streamer) session(ctx context.Context, resume bool) (res sessionResult) continue } res.healthy = true + if from, ok := s.replayFrom[frame.Channel]; ok && frame.Time > from { + s.replayFrom[frame.Channel] = frame.Time + } for _, data := range rows { s.noteBar(frame.Channel, data) if !s.Handlers.Row(StreamRow{Channel: frame.Channel, Replay: true, Data: data}) { @@ -443,24 +521,52 @@ func (s *Streamer) session(ctx context.Context, resume bool) (res sessionResult) return res } case frame.Op == "replayed": + since, follow, why := s.nextReplay(frame, stalled, continuations) if s.Handlers.Replayed != nil { s.Handlers.Replayed(StreamReplay{ Channel: frame.Channel, From: frame.From, To: frame.To, Bars: frame.Bars, Rows: frame.Rows, Missing: frame.Missing, Truncated: frame.Truncated, Unavailable: frame.Unavailable, + FollowedUp: follow, }) } - // The server stopped at its size cap: ask for the rest, from the - // bar after the last one sent. - step, replayable := streamReplaySteps[channelInterval(frame.Channel)] - if frame.Truncated && frame.To != nil && replayable && continuations < streamMaxContinuation { - if ch, ok := s.channelNamed(frame.Channel); ok { - continuations++ - ch.Since = *frame.To + step - if err := write(fmt.Sprintf("r%d", continuations), []StreamChannel{ch}); err != nil { - res.err = err - return res - } + if why != "" && s.Handlers.ServerErr != nil { + s.Handlers.ServerErr("replay_stopped", why) + } + if !follow { + delete(s.replayFrom, frame.Channel) + continue + } + if err := followUp(frame.Channel, since); err != nil { + res.err = err + return res + } + case frame.Op == "error" && frame.Code == "limit_exceeded" && followUps[frame.ID] != "": + // A follow-up refused while the server is busy: send it again + // after a jittered pause, then give the rest up to REST. + channel := followUps[frame.ID] + delete(followUps, frame.ID) + since, ok := s.replayFrom[channel] + if !ok { + continue + } + refusals++ + if refusals > streamBusyRetries { + delete(s.replayFrom, channel) + if s.Handlers.ServerErr != nil { + s.Handlers.ServerErr(frame.Code, fmt.Sprintf("the server stayed too busy to finish replaying %s; fill it from %s from the REST API", + channel, time.UnixMilli(since).UTC().Format(time.RFC3339))) } + continue + } + select { + case <-ctx.Done(): + res.err = ctx.Err() + return res + case <-time.After(streamBusyRetryBase + rand.N(streamBusyRetryBase+1)): + } + if err := followUp(channel, since); err != nil { + res.err = err + return res } case frame.Op == "error" && frame.ID == streamSubscribeID && frame.Code == "limit_exceeded" && hasSince(channels): // Too many replays at once on the server (as after it restarts): @@ -468,6 +574,7 @@ func (s *Streamer) session(ctx context.Context, resume bool) (res sessionResult) busyRetries++ if busyRetries > streamBusyRetries { channels = withoutSince(channels) + clear(s.replayFrom) if s.Handlers.ServerErr != nil { s.Handlers.ServerErr(frame.Code, "the server stayed too busy to replay; subscribing without it") } diff --git a/stream_replay_test.go b/stream_replay_test.go index 37c7eb2..81a821d 100644 --- a/stream_replay_test.go +++ b/stream_replay_test.go @@ -251,3 +251,109 @@ func TestStream_NoSinceBelowAMinuteOnReconnect(t *testing.T) { t.Errorf("expected no since at 15s, got %d", got) } } + +func emptyTruncatedFrame(id string, from int64) string { + return `{"op":"replayed","id":"` + id + `","channel":"` + streamTestChannel + `","from":` + strconv.FormatInt(from, 10) + + `,"to":null,"bars":0,"rows":0,"missing":[],"truncated":true}` +} + +// The size cap is shared by a subscribe's channels: one can leave another +// with no bars and no `to`. It is asked for again from `from`; empty a second +// time from the same bar, that bar is larger than the cap, and the stream +// stops asking. +func TestStream_EmptyTruncatedReplayIsAskedAgainThenGivenUp(t *testing.T) { + from := barMicros / 1000 + followUp := make(chan fakeSubscribe, 1) + startFakeStream(t, &fakeStream{ + script: func(ctx context.Context, conn *websocket.Conn, sub fakeSubscribe, _ int) { + ack(ctx, conn, sub, "") + send(ctx, conn, emptyTruncatedFrame(sub.ID, from)) + next := readSubscribe(ctx, t, conn) + followUp <- next + send(ctx, conn, emptyTruncatedFrame(next.ID, from)) + send(ctx, conn, row("2")) + drain(ctx, conn) + }, + }) + + _, stderr, code := runCLI("stream", "ohlcv", "--since", "1h", "--count", "1", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + if next := <-followUp; next.Channels[0].Since != from { + t.Errorf("expected the follow-up from the empty replay's from %d, got %+v", from, next) + } + if !strings.Contains(stderr, "larger than the server's replay cap") { + t.Errorf("expected the stop on stderr, got: %s", stderr) + } + if strings.Count(stderr, "more to come") != 1 { + t.Errorf("expected more to come only for the first replay, got: %s", stderr) + } +} + +// Held live rows come before `replayed`, so the newest bar received runs +// ahead of a replay still catching up; a drop then must not skip the rest. +func TestStream_DropDuringACatchUpResumesWhereTheReplayGotTo(t *testing.T) { + from := barMicros / 1000 + f := startFakeStream(t, &fakeStream{ + script: func(ctx context.Context, conn *websocket.Conn, sub fakeSubscribe, n int) { + ack(ctx, conn, sub, "") + if n == 1 { + send(ctx, conn, replayFrame(from, `{"symbol":"perpetual-BTC-USDT:USDT","time":1791307260000000}`)) + send(ctx, conn, `{"channel":"`+streamTestChannel+`","data":{"symbol":"perpetual-BTC-USDT:USDT","time":1791307860000000,"close":1}}`) + send(ctx, conn, replayedFrame(sub.ID, from, from, 1, 1, true)) + readSubscribe(ctx, t, conn) + conn.CloseNow() + return + } + send(ctx, conn, row("3")) + drain(ctx, conn) + }, + }) + + _, stderr, code := runCLI("stream", "ohlcv", "--since", "1h", "--count", "2", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + _, _, subs := f.recorded() + if len(subs) != 2 { + t.Fatalf("expected two subscribes, got %d", len(subs)) + } + if got := subs[1].Channels[0].Since; got != from+60_000 { + t.Errorf("expected the reconnect to resume the replay from %d, got %d", from+60_000, got) + } +} + +func TestStream_RefusedFollowUpIsSentAgain(t *testing.T) { + base := streamBusyRetryBase + streamBusyRetryBase = 10 * time.Millisecond + t.Cleanup(func() { streamBusyRetryBase = base }) + + from := barMicros / 1000 + followUps := make(chan fakeSubscribe, 2) + startFakeStream(t, &fakeStream{ + script: func(ctx context.Context, conn *websocket.Conn, sub fakeSubscribe, _ int) { + ack(ctx, conn, sub, "") + send(ctx, conn, replayFrame(from, `{"symbol":"perpetual-BTC-USDT:USDT","time":1791307260000000}`)) + send(ctx, conn, replayedFrame(sub.ID, from, from, 1, 1, true)) + first := readSubscribe(ctx, t, conn) + followUps <- first + send(ctx, conn, `{"op":"error","id":"`+first.ID+`","code":"limit_exceeded","message":"At most 4 replays can be queued on a connection"}`) + followUps <- readSubscribe(ctx, t, conn) + send(ctx, conn, row("2")) + drain(ctx, conn) + }, + }) + + _, stderr, code := runCLI("stream", "ohlcv", "--since", "1h", "--count", "1", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + first, second := <-followUps, <-followUps + if first.Channels[0].Since != from+60_000 || second.Channels[0].Since != from+60_000 { + t.Errorf("expected the refused follow-up sent again from %d, got %+v then %+v", from+60_000, first, second) + } + if strings.Contains(stderr, "limit_exceeded") { + t.Errorf("expected the refusal handled, not warned, got: %s", stderr) + } +}