fix(Stream): Resume an unfinished replay, ask again for an empty one and retry refused follow-ups

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;

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)

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,31 +521,60 @@ 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):
 			// try again after a jittered pause, then without the replay.
 			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")
 				}

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)
+	}
+}