From cf38a3aeb9ff9997adea84045d20263e3bb79298 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 9 Nov 2026 18:05:53 +0000 Subject: [PATCH] feature(Stream): Replay missed bars on reconnect, add --since and --replay Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011HGNUuQYbp3ZUdqW8UY5Kf --- README.md | 12 +- cli_stream.go | 67 ++++++++++- stream.go | 238 +++++++++++++++++++++++++++++++++++++-- stream_replay_test.go | 253 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 551 insertions(+), 19 deletions(-) create mode 100644 stream_replay_test.go diff --git a/README.md b/README.md index f804f88..3d8b02d 100644 --- a/README.md +++ b/README.md @@ -289,7 +289,8 @@ 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. Delivery is at most once: rows published while disconnected are not replayed. +- 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. - 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; @@ -303,8 +304,15 @@ aperiodic stream ohlcv \ | `--exchange` | Exchange name | `binance-futures` | | `--interval` | Aggregation interval | `1m` | | `--symbols` | Comma-separated symbols | every symbol your plan allows | -| `--count` | Stop after this many live rows (snapshot rows are printed but not counted) | no limit | +| `--count` | Stop after this many live rows (snapshot and replayed rows are printed but not counted) | no limit | | `--duration` | Stop after this long, e.g. `90s`, `1h` | no limit | +| `--since` | Replay the bars from this time on before the live rows: an RFC 3339 time (`2026-11-09T10:00:00Z`) or how long ago (`30m`). 1m and longer intervals | none | +| `--replay` | After a reconnect, replay the bars missed while disconnected | `true` | + +**Catch up on the last 30 minutes, then keep streaming:** +```bash +aperiodic stream ohlcv --symbols perpetual-BTC-USDT:USDT --since 30m +``` **Take the next closed 1m bar and stop:** ```bash diff --git a/cli_stream.go b/cli_stream.go index 97db0a0..f7e1bcc 100644 --- a/cli_stream.go +++ b/cli_stream.go @@ -20,6 +20,8 @@ type streamFlags struct { symbols string count int duration time.Duration + since string + replay bool } func newStreamFlagSet() (*flag.FlagSet, *streamFlags) { @@ -28,11 +30,28 @@ func newStreamFlagSet() (*flag.FlagSet, *streamFlags) { fs.StringVar(&f.exchange, "exchange", string(ExchangeBinanceFutures), "Exchange name") fs.StringVar(&f.interval, "interval", string(Interval1m), "Aggregation interval") fs.StringVar(&f.symbols, "symbols", "", "Comma-separated symbols, e.g. perpetual-BTC-USDT:USDT (default: every symbol your plan allows)") - fs.IntVar(&f.count, "count", 0, "Stop after this many live rows; snapshot rows are printed but not counted (0: no limit)") + fs.IntVar(&f.count, "count", 0, "Stop after this many live rows; snapshot and replayed rows are printed but not counted (0: no limit)") fs.DurationVar(&f.duration, "duration", 0, "Stop after this long, e.g. 90s or 1h (0: no limit)") + fs.StringVar(&f.since, "since", "", "Replay the bars from this time on before the live rows: an RFC 3339 time, e.g. 2026-11-09T10:00:00Z, or how long ago, e.g. 30m (1m and longer intervals)") + fs.BoolVar(&f.replay, "replay", true, "After a reconnect, replay the bars missed while disconnected (1m and longer intervals)") return fs, f } +// parseSince reads --since: an RFC 3339 time, or a duration back from now. +func parseSince(value string, now time.Time) (time.Time, error) { + if value == "" { + return time.Time{}, nil + } + if at, err := time.Parse(time.RFC3339, value); err == nil { + return at, nil + } + ago, err := time.ParseDuration(value) + if err != nil || ago <= 0 { + return time.Time{}, fmt.Errorf("--since must be an RFC 3339 time or a positive duration such as 30m, got %q", value) + } + return now.Add(-ago), nil +} + func (c *CLI) runStream(args []string) int { if len(args) > 0 && isHelpArg(args[0]) { c.printStreamUsage() @@ -59,6 +78,17 @@ func (c *CLI) runStream(args []string) int { fmt.Fprintln(c.Stderr, "Error: --duration must not be negative") return 1 } + since, err := parseSince(f.since, time.Now()) + if err != nil { + fmt.Fprintf(c.Stderr, "Error: %v\n", err) + return 1 + } + if !since.IsZero() { + if _, ok := streamReplaySteps[f.interval]; !ok { + fmt.Fprintf(c.Stderr, "Error: --since needs an interval of 1m or longer, got %s\n", f.interval) + return 1 + } + } apiKey := c.Env("APERIODIC_API_KEY") if apiKey == "" { @@ -102,7 +132,7 @@ func (c *CLI) runStream(args []string) int { fmt.Fprintf(c.Stderr, "Warning: skipping a row: %v\n", err) return true } - if !row.Snapshot { + if !row.Snapshot && !row.Replay { live++ } return f.count == 0 || live < f.count @@ -123,6 +153,23 @@ func (c *CLI) runStream(args []string) int { fmt.Fprintf(c.Stderr, "Rejected: %s (%s): %s\n", r.Channel, r.Code, r.Message) } }, + Replayed: func(r StreamReplay) { + if r.Unavailable { + fmt.Fprintf(c.Stderr, "Replay unavailable for %s: fill the gap from the REST API\n", r.Channel) + return + } + fmt.Fprintf(c.Stderr, "Replayed %s: %d bars, %d rows", r.Channel, r.Bars, r.Rows) + if r.Bars > 0 && r.To != nil { + fmt.Fprintf(c.Stderr, " (%s to %s)", formatBarTime(r.From), formatBarTime(*r.To)) + } + if len(r.Missing) > 0 { + fmt.Fprintf(c.Stderr, "; %d bars the server does not hold", len(r.Missing)) + } + if r.Truncated { + fmt.Fprint(c.Stderr, "; more to come") + } + fmt.Fprintln(c.Stderr) + }, ServerErr: func(code, message string) { fmt.Fprintf(c.Stderr, "Warning: server error %s: %s\n", code, message) }, @@ -135,7 +182,10 @@ func (c *CLI) runStream(args []string) int { }, }) - err := streamer.Run(ctx) + streamer.Since = since + streamer.Replay = f.replay + + err = streamer.Run(ctx) if sig, ok := interrupted.Load().(os.Signal); ok { return signalExitCode(sig) } @@ -153,6 +203,10 @@ func (c *CLI) runStream(args []string) int { return 1 } +func formatBarTime(ms int64) string { + return time.UnixMilli(ms).UTC().Format(time.RFC3339) +} + // signalExitCode follows the shell convention of 128 + the signal number. func signalExitCode(sig os.Signal) int { if sig == syscall.SIGTERM { @@ -172,8 +226,11 @@ func (c *CLI) printStreamUsage() { fmt.Fprintln(out, ` {"channel":"ohlcv.binance-futures.1m","snapshot":false,"data":{...}}`) fmt.Fprintln(out, "Snapshot rows (the latest row per symbol, sent on subscribe on some plans)") fmt.Fprintln(out, "have \"snapshot\":true. Subscriptions, rejections, reconnects and warnings go") - fmt.Fprintln(out, "to stderr. Delivery is at most once: rows missed while reconnecting are not") - fmt.Fprintln(out, "replayed. Exits 1 if the key is refused, the plan has no live access, the") + fmt.Fprintln(out, "to stderr. After a reconnect, the bars missed meanwhile are replayed first") + fmt.Fprintln(out, "(1m and longer intervals, as far back as the server keeps them), marked") + fmt.Fprintln(out, "\"replay\":true; --since replays from a time of your choosing. A row can") + fmt.Fprintln(out, "come twice; it has the same key, so overwrite it. Exits 1 if the key is") + fmt.Fprintln(out, "refused, the plan has no live access, the") fmt.Fprintln(out, "connection limit is reached on the first connect, no channel is (or stays)") fmt.Fprintln(out, "subscribed, or the server ends the session for good. Ctrl-C exits 130,") fmt.Fprintln(out, "SIGTERM 143.") diff --git a/stream.go b/stream.go index 7e72a0a..6dc87a0 100644 --- a/stream.go +++ b/stream.go @@ -23,17 +23,34 @@ const ( // The server sends a heartbeat every ~30s when idle; a silent socket // for longer than this is treated as dropped. - streamIdleTimeout = 75 * time.Second - streamDialTimeout = 15 * time.Second - streamMessageLimit = 1 << 20 + 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. + streamMessageLimit = 16 << 20 + + // How often a subscribe refused because the server is busy replaying is + // sent again with its since, and how many truncated replays are followed + // up, before giving up on the replay. + streamBusyRetries = 5 + streamMaxContinuation = 50 ) var ( streamBackoffBase = 500 * time.Millisecond streamBackoffMax = 30 * time.Second streamHealthyAfter = 60 * time.Second + // A busy refusal is retried after this, plus as much again at random. + streamBusyRetryBase = time.Second ) +// streamReplaySteps are the intervals the server keeps bars for, with their +// length in ms. A since on any other interval is refused. +var streamReplaySteps = map[string]int64{ + "1m": 60_000, "5m": 300_000, "15m": 900_000, "30m": 1_800_000, + "1h": 3_600_000, "4h": 14_400_000, "1d": 86_400_000, +} + // StreamChannel is one subscription; Symbols nil means every symbol the // plan allows. type StreamChannel struct { @@ -41,6 +58,13 @@ type StreamChannel struct { Exchange string `json:"exchange"` Interval string `json:"interval"` Symbols []string `json:"symbols,omitempty"` + // Since asks for the channel's bars from this time on (epoch ms) before + // its live rows; 0 asks for none. + Since int64 `json:"since,omitempty"` +} + +func (c StreamChannel) name() string { + return c.Dataset + "." + c.Exchange + "." + c.Interval } type StreamRejection struct { @@ -49,13 +73,28 @@ type StreamRejection struct { Message string `json:"message"` } -// StreamRow is a data frame, printed as one JSON line. +// StreamRow is a data frame, printed as one JSON line. Replay marks a row +// from the history sent before the live rows. type StreamRow struct { Channel string `json:"channel"` Snapshot bool `json:"snapshot"` + Replay bool `json:"replay,omitempty"` Data json.RawMessage `json:"data"` } +// StreamReplay is the server's account of one channel's replay. Times are +// bar open times in epoch ms; Unavailable means the history could not be read. +type StreamReplay struct { + Channel string + From int64 + To *int64 + Bars int + Rows int + Missing []int64 + Truncated bool + Unavailable bool +} + type streamFrame struct { Op string `json:"op"` ID string `json:"id"` @@ -66,7 +105,16 @@ type streamFrame struct { Channel string `json:"channel"` Snapshot bool `json:"snapshot"` + Replay bool `json:"replay"` Data json.RawMessage `json:"data"` + + From int64 `json:"from"` + To *int64 `json:"to"` + Bars int `json:"bars"` + Rows int `json:"rows"` + Missing []int64 `json:"missing"` + Truncated bool `json:"truncated"` + Unavailable bool `json:"unavailable"` } // StreamHandshakeError is an HTTP refusal before the upgrade. @@ -118,6 +166,7 @@ type StreamHandlers struct { Row func(StreamRow) bool Subscribed func(granted []string, rejected []StreamRejection) Unsubscribed func(removed []string, rejected []StreamRejection) + Replayed func(StreamReplay) ServerErr func(code, message string) // Reconnect reports a retry; connected says whether the failed attempt // had a connection to lose. @@ -129,6 +178,18 @@ type Streamer struct { Header http.Header Channels []StreamChannel Handlers StreamHandlers + // Since asks for every channel's bars from this time on before its live + // rows, on the first connect (zero: none). + Since time.Time + // Replay asks, on every reconnect, for the bars of each channel since the + // newest one received, so a reconnect leaves no gap the server still holds. + Replay bool + + // lastBar is each channel's newest bar time received (epoch ms), and + // subscribedAt when the last session subscribed: where a reconnect's + // replay starts. + lastBar map[string]int64 + subscribedAt time.Time } // NewStreamer reads APERIODIC_STREAM_URL and the Cloudflare Access service @@ -146,7 +207,62 @@ func NewStreamer(apiKey string, env func(string) string, channels []StreamChanne url = DefaultStreamURL } - return &Streamer{URL: url, Header: header, Channels: channels, Handlers: handlers} + return &Streamer{URL: url, Header: header, Channels: channels, Handlers: handlers, Replay: true} +} + +// subscribeChannels is the channels to subscribe with, each with the since it +// should replay from, if any: Since on the first session, and on a reconnect +// the newest bar received (or, when none came, one bar before the last +// subscribe). Intervals the server keeps no history for get none. +func (s *Streamer) subscribeChannels(resume bool) []StreamChannel { + channels := make([]StreamChannel, len(s.Channels)) + for i, ch := range s.Channels { + channels[i] = ch + step, ok := streamReplaySteps[ch.Interval] + if !ok || ch.Since != 0 && !resume { + continue + } + switch { + case !resume && !s.Since.IsZero(): + channels[i].Since = s.Since.UnixMilli() + case resume && s.Replay && s.lastBar[ch.name()] > 0: + channels[i].Since = s.lastBar[ch.name()] + case resume && s.Replay && !s.subscribedAt.IsZero(): + channels[i].Since = s.subscribedAt.UnixMilli() - step + default: + channels[i].Since = 0 + } + } + return channels +} + +// 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 { + Time json.Number `json:"time"` + } + if json.Unmarshal(data, &row) != nil { + return + } + value, err := row.Time.Float64() + if err != nil || value <= 0 { + return + } + ms := int64(value) + switch { + case value >= 1e17: + ms = int64(value / 1e6) + case value >= 1e14: + ms = int64(value / 1e3) + case value < 1e11: + ms = int64(value * 1e3) + } + if s.lastBar == nil { + s.lastBar = map[string]int64{} + } + if ms > s.lastBar[channel] { + s.lastBar[channel] = ms + } } // sessionResult describes how one connection went. @@ -167,7 +283,7 @@ func (s *Streamer) Run(ctx context.Context) error { attempt := 0 everConnected, everSubscribed := false, false for { - res := s.session(ctx) + res := s.session(ctx, everSubscribed) if errors.Is(res.err, errStreamDone) || ctx.Err() != nil { return nil } @@ -233,7 +349,7 @@ func streamBackoff(attempt int) time.Duration { return half + rand.N(half+1) } -func (s *Streamer) session(ctx context.Context) (res sessionResult) { +func (s *Streamer) session(ctx context.Context, resume bool) (res sessionResult) { conn, err := s.dial(ctx) if err != nil { var handshake *StreamHandshakeError @@ -259,17 +375,22 @@ func (s *Streamer) session(ctx context.Context) (res sessionResult) { }) defer stop() - subscribe, err := json.Marshal(map[string]any{"op": "subscribe", "id": streamSubscribeID, "channels": s.Channels}) - if err != nil { - res.err = err - return res + channels := s.subscribeChannels(resume) + write := func(id string, channels []StreamChannel) error { + frame, err := json.Marshal(map[string]any{"op": "subscribe", "id": id, "channels": channels}) + if err != nil { + return err + } + return conn.Write(ctx, websocket.MessageText, frame) } - if err := conn.Write(ctx, websocket.MessageText, subscribe); err != nil { + if err := write(streamSubscribeID, channels); err != nil { res.err = err return res } + s.subscribedAt = time.Now() granted := map[string]bool{} + busyRetries, continuations := 0, 0 for { msg, err := readStreamMessage(conn) if err != nil { @@ -293,14 +414,75 @@ func (s *Streamer) session(ctx context.Context) (res sessionResult) { } switch { + case frame.Replay && frame.Channel != "": + var rows []json.RawMessage + if err := json.Unmarshal(frame.Data, &rows); err != nil { + if s.Handlers.ServerErr != nil { + s.Handlers.ServerErr("unreadable_frame", truncate(string(msg), 200)) + } + continue + } + res.healthy = true + for _, data := range rows { + s.noteBar(frame.Channel, data) + if !s.Handlers.Row(StreamRow{Channel: frame.Channel, Replay: true, Data: data}) { + _ = conn.Close(websocket.StatusNormalClosure, "") + res.err = errStreamDone + return res + } + } case frame.Channel != "" && frame.Data != nil: res.healthy = true + if !frame.Snapshot { + s.noteBar(frame.Channel, frame.Data) + } row := StreamRow{Channel: frame.Channel, Snapshot: frame.Snapshot, Data: frame.Data} if !s.Handlers.Row(row) { _ = conn.Close(websocket.StatusNormalClosure, "") res.err = errStreamDone return res } + case frame.Op == "replayed": + 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, + }) + } + // 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 + } + } + } + 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) + if s.Handlers.ServerErr != nil { + s.Handlers.ServerErr(frame.Code, "the server stayed too busy to replay; subscribing without it") + } + } else { + select { + case <-ctx.Done(): + res.err = ctx.Err() + return res + case <-time.After(streamBusyRetryBase + rand.N(streamBusyRetryBase+1)): + } + } + if err := write(streamSubscribeID, channels); err != nil { + res.err = err + return res + } case frame.Op == "subscribed": res.subscribed = true for _, ch := range frame.Channels { @@ -387,3 +569,35 @@ func truncate(s string, n int) string { } return s[:n] + "..." } + +func (s *Streamer) channelNamed(name string) (StreamChannel, bool) { + for _, ch := range s.Channels { + if ch.name() == name { + return ch, true + } + } + return StreamChannel{}, false +} + +func channelInterval(name string) string { + parts := strings.Split(name, ".") + return parts[len(parts)-1] +} + +func hasSince(channels []StreamChannel) bool { + for _, ch := range channels { + if ch.Since != 0 { + return true + } + } + return false +} + +func withoutSince(channels []StreamChannel) []StreamChannel { + out := make([]StreamChannel, len(channels)) + for i, ch := range channels { + ch.Since = 0 + out[i] = ch + } + return out +} diff --git a/stream_replay_test.go b/stream_replay_test.go new file mode 100644 index 0000000..37c7eb2 --- /dev/null +++ b/stream_replay_test.go @@ -0,0 +1,253 @@ +package aperiodic + +import ( + "context" + "encoding/json" + "strconv" + "strings" + "testing" + "time" + + "github.com/coder/websocket" +) + +// barMicros is the bar time the fake rows carry (row in stream_test.go), in +// the microseconds the topics use. +const barMicros int64 = 1791307260000000 + +type replayLine struct { + Channel string `json:"channel"` + Snapshot bool `json:"snapshot"` + Replay bool `json:"replay"` + Data json.RawMessage `json:"data"` +} + +func parseReplayLines(t *testing.T, stdout string) []replayLine { + t.Helper() + var lines []replayLine + for _, raw := range strings.Split(strings.TrimSpace(stdout), "\n") { + if raw == "" { + continue + } + var line replayLine + if err := json.Unmarshal([]byte(raw), &line); err != nil { + t.Fatalf("stdout line is not JSON: %q", raw) + } + lines = append(lines, line) + } + return lines +} + +func replayFrame(time int64, rows ...string) string { + return `{"channel":"` + streamTestChannel + `","replay":true,"time":` + strconv.FormatInt(time, 10) + `,"data":[` + strings.Join(rows, ",") + `]}` +} + +func replayedFrame(id string, from, to int64, bars, rows int, truncated bool) string { + return `{"op":"replayed","id":"` + id + `","channel":"` + streamTestChannel + `","from":` + strconv.FormatInt(from, 10) + + `,"to":` + strconv.FormatInt(to, 10) + `,"bars":` + strconv.Itoa(bars) + `,"rows":` + strconv.Itoa(rows) + + `,"missing":[],"truncated":` + strconv.FormatBool(truncated) + `}` +} + +func readSubscribe(ctx context.Context, t *testing.T, conn *websocket.Conn) fakeSubscribe { + t.Helper() + _, msg, err := conn.Read(ctx) + if err != nil { + t.Errorf("expected another subscribe: %v", err) + return fakeSubscribe{} + } + var sub fakeSubscribe + if err := json.Unmarshal(msg, &sub); err != nil { + t.Errorf("subscribe is not JSON: %s", msg) + } + return sub +} + +// A replay frame holds a whole bar, over a megabyte on an all-symbols channel: +// a client that cannot read one reconnects and asks for it again forever. +func TestStream_SinceReplaysBeforeTheLiveRows(t *testing.T) { + from := time.Date(2026, 11, 9, 10, 0, 0, 0, time.UTC).UnixMilli() + padding := strings.Repeat("x", 2<<20) + f := 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":1791307200000000,"close":1}`, + `{"symbol":"perpetual-ETH-USDT:USDT","time":1791307200000000,"pad":"`+padding+`"}`)) + send(ctx, conn, replayedFrame(sub.ID, from, from, 1, 2, false)) + send(ctx, conn, row("3")) + drain(ctx, conn) + }, + }) + + stdout, stderr, code := runCLI("stream", "ohlcv", "--since", "2026-11-09T10:00:00Z", "--count", "1", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + + _, _, subs := f.recorded() + if got := subs[0].Channels[0].Since; got != from { + t.Errorf("expected since %d, got %d", from, got) + } + lines := parseReplayLines(t, stdout) + if len(lines) != 3 || !lines[0].Replay || !lines[1].Replay || lines[2].Replay { + t.Fatalf("expected two replayed rows then a live one, got: %.300s", stdout) + } + if !strings.Contains(stderr, "Replayed "+streamTestChannel+": 1 bars, 2 rows") { + t.Errorf("expected the replay summary on stderr, got: %s", stderr) + } +} + +func TestStream_ReconnectReplaysFromTheNewestBar(t *testing.T) { + 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, row("1")) + conn.CloseNow() + return + } + send(ctx, conn, replayFrame(barMicros/1000, `{"symbol":"perpetual-BTC-USDT:USDT","time":1791307320000000,"close":2}`)) + send(ctx, conn, replayedFrame(sub.ID, barMicros/1000, barMicros/1000+60_000, 1, 1, false)) + send(ctx, conn, row("3")) + drain(ctx, conn) + }, + }) + + stdout, stderr, code := runCLI("stream", "ohlcv", "--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[0].Channels[0].Since; got != 0 { + t.Errorf("expected no since on the first connect, got %d", got) + } + if got := subs[1].Channels[0].Since; got != barMicros/1000 { + t.Errorf("expected the reconnect to replay from the newest bar %d, got %d", barMicros/1000, got) + } + lines := parseReplayLines(t, stdout) + if len(lines) != 3 || lines[0].Replay || !lines[1].Replay || lines[2].Replay { + t.Errorf("expected live, replayed, live, got: %s", stdout) + } +} + +func TestStream_NoReplayAfterAReconnectWhenTurnedOff(t *testing.T) { + 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, row("1")) + conn.CloseNow() + return + } + send(ctx, conn, row("2")) + drain(ctx, conn) + }, + }) + + _, stderr, code := runCLI("stream", "ohlcv", "--replay=false", "--count", "2", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + _, _, subs := f.recorded() + if got := subs[1].Channels[0].Since; got != 0 { + t.Errorf("expected no since with --replay=false, got %d", got) + } +} + +// After a deploy every client reconnects with since at once; the server +// refuses past its cap and expects a retry after a pause. +func TestStream_BusyRefusalIsSentAgain(t *testing.T) { + base := streamBusyRetryBase + streamBusyRetryBase = 10 * time.Millisecond + t.Cleanup(func() { streamBusyRetryBase = base }) + + retried := make(chan fakeSubscribe, 1) + startFakeStream(t, &fakeStream{ + script: func(ctx context.Context, conn *websocket.Conn, sub fakeSubscribe, _ int) { + send(ctx, conn, `{"op":"error","id":"`+sub.ID+`","code":"limit_exceeded","message":"Too many replays are running at once"}`) + again := readSubscribe(ctx, t, conn) + retried <- again + ack(ctx, conn, again, "") + send(ctx, conn, row("1")) + drain(ctx, conn) + }, + }) + + _, stderr, code := runCLI("stream", "ohlcv", "--since", "30m", "--count", "1", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + if again := <-retried; again.ID != streamSubscribeID || again.Channels[0].Since == 0 { + t.Errorf("expected the same subscribe, since included, sent again; got %+v", again) + } +} + +// The server caps one replay; the rest comes from asking again after it. +func TestStream_TruncatedReplayIsFollowedUp(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, replayFrame(from, `{"symbol":"perpetual-BTC-USDT:USDT","time":1791307260000000}`)) + send(ctx, conn, replayedFrame(sub.ID, from, from, 1, 1, true)) + next := readSubscribe(ctx, t, conn) + followUp <- next + 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) + } + next := <-followUp + if next.ID == streamSubscribeID || len(next.Channels) != 1 || next.Channels[0].Since != from+60_000 { + t.Errorf("expected a follow-up subscribe from the next bar, got %+v", next) + } + if !strings.Contains(stderr, "more to come") { + t.Errorf("expected the truncation on stderr, got: %s", stderr) + } +} + +// The server keeps no history below a minute and refuses since there. +func TestStream_SinceNeedsAMinuteOrLonger(t *testing.T) { + t.Setenv("APERIODIC_API_KEY", streamTestKey) + _, stderr, code := runCLI("stream", "ohlcv", "--interval", "15s", "--since", "30m") + if code != 1 || !strings.Contains(stderr, "1m or longer") { + t.Fatalf("expected exit 1 naming the interval, got %d: %s", code, stderr) + } + _, stderr, code = runCLI("stream", "ohlcv", "--since", "yesterday") + if code != 1 || !strings.Contains(stderr, "RFC 3339") { + t.Fatalf("expected exit 1 explaining --since, got %d: %s", code, stderr) + } +} + +func TestStream_NoSinceBelowAMinuteOnReconnect(t *testing.T) { + 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, row("1")) + conn.CloseNow() + return + } + send(ctx, conn, row("2")) + drain(ctx, conn) + }, + }) + + _, stderr, code := runCLI("stream", "ohlcv", "--interval", "15s", "--count", "2", "--duration", "10s") + if code != 0 { + t.Fatalf("expected exit code 0, got %d; stderr: %s", code, stderr) + } + _, _, subs := f.recorded() + if got := subs[1].Channels[0].Since; got != 0 { + t.Errorf("expected no since at 15s, got %d", got) + } +}