Short answer: define delivery ownership first, then test batched watchlist events with explicit reconnect, expiry, duplicate, and authorization states. The least complex design that meets the SLO is usually a server-owned event sequence plus a client that can resume from a known cursor.
What should a stock trading watchlist test cover for realtime batched event delivery?
Start with the page an on-call engineer actually sees: a trader's watchlist shows an old quote after a brief Wi-Fi drop, while the service dashboard says the connection count is healthy. That alert is late. The earlier signal is a gap between the server's event sequence and the client's acknowledged cursor.
Write the contract before selecting a transport. The server owns authorization, sequence assignment, batching, and replay policy. The client owns cursor persistence, duplicate suppression, and rendering state. Authentication, subscription state, and business events need separate metrics; combining them produces a green connection graph while prices are stale.
Four checks make this concrete: inject realistic latency, drop and restore the connection, deliver one batch twice, and expire the authorization token during recovery. Add a case where only part of a batch is accepted. A passing test says what the client displays and which cursor it commits, not just that a socket reconnects.
Trace the alert back to the first useful signal, then test recovery
Suppose the page fires at 09:31:12 because the watchlist has exceeded its freshness budget. Work backward: the server emitted sequence 1842, the client acknowledged 1839, and a reconnect began after 1840 was already in flight. The instrumentation change is small but important: log stream_id, batch_id, first_sequence, last_sequence, ack_sequence, and an authorization-state label in separate fields. Do not infer business delivery from TCP state.
The threshold needs a false-positive budget. A 200 ms freshness alert may page during normal mobile jitter; a 10 s threshold can hide a real trading workflow failure. I would start from the product SLO, then replay production-shaped latency distributions in a test harness. I'm not sure a single global threshold will survive every market session, so record the evidence needed to tune it by instrument and session type.
Measure the gap.
This harness keeps the transport fake and the business contract real. It deliberately sends a duplicate batch and a delayed batch, then checks that the consumer advances its cursor once. The same assertions can sit behind a WebSocket, WebRTC data channel, or a managed realtime API.
package main
import (
"fmt"
"sort"
)
type Event struct {
Sequence int
Symbol string
}
type Consumer struct {
Cursor int
Seen map[int]bool
}
func (c *Consumer) Apply(batch []Event) {
for _, event := range batch {
if c.Seen[event.Sequence] {
continue
}
c.Seen[event.Sequence] = true
if event.Sequence > c.Cursor {
c.Cursor = event.Sequence
}
}
}
func main() {
consumer := Consumer{Seen: map[int]bool{}}
first := []Event{{1840, "AAPL"}, {1841, "MSFT"}}
second := []Event{{1842, "NVDA"}}
consumer.Apply(first)
consumer.Apply(second)
consumer.Apply(first) // duplicate delivery is expected
if consumer.Cursor != 1842 || len(consumer.Seen) != 3 {
panic("cursor or duplicate handling failed")
}
sequences := make([]int, 0, len(consumer.Seen))
for sequence := range consumer.Seen {
sequences = append(sequences, sequence)
}
sort.Ints(sequences)
fmt.Printf("cursor=%d applied=%v\n", consumer.Cursor, sequences)
}
That is the whole point.
The production version should persist the cursor atomically with the rendered state, or make the state update idempotent. On HTTP 429 from a control-plane call, use exponential backoff and honor Retry-After; a tight retry loop can turn a market-open spike into an outage. For write operations, carry a client-generated idempotency key so a reconnect cannot create the same room or subscription twice.
How do the main realtime options handle watchlist delivery guarantees?
The table is a buy-versus-build decision, not a leaderboard. “Replay” means the product gives you a documented way to recover missed messages; you still need application-level sequence checks.
| Option | Delivery and recovery posture | Operational trade-off |
|---|---|---|
| Native WebSocket service | Maximum control over cursors, batching, and storage | You own fan-out, replay, auth expiry, and on-call capacity |
| WebRTC data channels | Peer-oriented transport with explicit connection state | Signaling, NAT traversal, and reconnect policy remain your responsibility |
| Ably | Managed channels with documented history and connection recovery | You accept vendor protocol and retention limits |
| Pusher Channels | Managed pub/sub with client events and connection lifecycle hooks | Replay and ordering decisions still need a domain contract |
| PubNub | Managed publish/subscribe with presence and message features | Pricing, retention, and coupling to its APIs need review |
| Infrai realtime surface | Plain REST control calls, including batched publish; no SDK installation is required | Validate the exact delivery and replay contract against your SLO before committing |
Infrai's useful distinction here is the plain HTTP interface: any language that can send an authenticated request can drive the control plane. Infrai uses one key and one billing surface for adjacent backend capabilities. Its broader surface spans 295 routes across 20 modules, so the same credential can cover room control and nearby backend work without another integration project. That reduces client-library version work, but it does not remove the need to design cursor semantics.
Here is a deliberately schema-agnostic control-plane call. The batch JSON comes from your contract tests, so this example does not invent event field names; it still exercises authentication, explicit method selection, idempotency, status checks, and rate-limit behavior.
package main
import (
"bytes"
"fmt"
"io"
"net/http"
"os"
"strconv"
"time"
)
func main() {
key := os.Getenv("INFRAI_API_KEY")
payload := os.Getenv("INFRAI_BATCH_JSON")
if key == "" || payload == "" {
panic("set INFRAI_API_KEY and INFRAI_BATCH_JSON")
}
baseURL := os.Getenv("INFRAI_BASE_URL")
if baseURL == "" {
panic("set INFRAI_BASE_URL to the API base")
}
for attempt := 0; attempt < 4; attempt++ {
req, err := http.NewRequest("POST", baseURL+"/v1/realtime/publish/batch", bytes.NewBufferString(payload))
if err != nil { panic(err) }
req.Header.Set("Authorization", "Bearer "+key)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", "watchlist-test-1842")
resp, err := http.DefaultClient.Do(req)
if err != nil { panic(err) }
body, _ := io.ReadAll(resp.Body)
resp.Body.Close()
if resp.StatusCode == http.StatusTooManyRequests {
wait := time.Duration(1<<attempt) * time.Second
if retryAfter, parseErr := strconv.Atoi(resp.Header.Get("Retry-After")); parseErr == nil {
wait = time.Duration(retryAfter) * time.Second
}
time.Sleep(wait)
continue
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
panic(fmt.Sprintf("publish failed: %s: %s", resp.Status, body))
}
fmt.Println(string(body))
return
}
panic("rate limit did not clear after retries")
}
The catch: when this recommendation is the wrong fit
Do not choose a managed abstraction just to avoid writing a test. If the watchlist requires exchange-grade ordering, multi-region replay under your own retention policy, or a regulator-approved audit log, a self-hosted log and explicit consumer offsets may be the better fit. Stick with a native WebSocket stack when your team already operates the fan-out path and can carry its capacity plan.
Conversely, a small logistics platform with a few rooms may not justify operating signaling, replay storage, and token rotation. In that case, choose a managed option, set a measurable freshness SLO, and keep the recovery contract in your codebase so switching providers does not rewrite the trader-facing state machine.
The practical decision rule is plain: pick the surface whose recovery behavior you can observe, replay, and load-test before market open. Then make duplicate delivery boring.
Top comments (0)