From 58e6addbabb802c0b39e10ed435cb2e9c5f01714 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Fri, 8 May 2026 13:21:51 +0200 Subject: [PATCH] trickle: align Go publisher with SDK probe-and-fallback pattern MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror the Python SDK's first-POST flow in the Go publisher: GET {baseURL}/next to learn the server's nextWrite, fall back to slot 0 on probe failure. Establishes a single canonical pattern across the two trickle clients in the Livepeer stack. Wire behavior on a fresh channel is unchanged — both old (always-0) and new (probe-then-resolve) paths POST to /0 first, then /1, /2, ... The probe adds one round-trip on session start; against a server with the /next route (livepeer/go-livepeer#3925) it returns nextWrite cleanly, against a server without it (today's master) the failure path resolves to 0 — the same starting slot. The fallback explicitly returns 0 (not -1) for the same reason the Python SDK does: -1 would let the server resolve to nextWrite, but the publisher has no read-back path to learn what slot the server picked. The local counter then increments past whatever the server chose, causing the second POST to race the first at slot 0. Returning 0 keeps the publisher's local counter in lockstep with the server. Tests in trickle_test.go cover both paths via mock /next handlers (success returning Lp-Trickle-Latest, fallback returning 400). The existing test suite continues to pass against the configured trickle server (which doesn't have /next on master) since the fallback path is the existing behavior in disguise. Refs livepeer/go-livepeer#3925, livepeer/livepeer-python-gateway#12. --- trickle/trickle_publisher.go | 35 +++++++++++++++- trickle/trickle_test.go | 81 ++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 1 deletion(-) diff --git a/trickle/trickle_publisher.go b/trickle/trickle_publisher.go index 13d5479e98..9f82e44f30 100644 --- a/trickle/trickle_publisher.go +++ b/trickle/trickle_publisher.go @@ -7,6 +7,7 @@ import ( "io" "log/slog" "net/http" + "strconv" "sync" "time" ) @@ -44,12 +45,18 @@ type pendingPost struct { client *TricklePublisher } -// NewTricklePublisher creates a new trickle stream client +// NewTricklePublisher creates a new trickle stream client. Probes /next +// for the next-write slot; falls back to 0 when /next is unavailable. func NewTricklePublisher(url string) (*TricklePublisher, error) { c := &TricklePublisher{ baseURL: url, contentType: "video/MP2T", client: httpClient(), + index: -1, // sentinel: resolved before first POST. Mirrors the + // subscriber's `Next = -1` and the Python SDK's start_seq=-1. + } + if c.index < 0 { + c.index = c.resolveNextSeq() } p, err := c.preconnect() if err != nil { @@ -60,6 +67,32 @@ func NewTricklePublisher(url string) (*TricklePublisher, error) { return c, nil } +// resolveNextSeq probes /next for the server's next-write slot. Returns 0 +// on any failure — returning -1 would race the server's slot resolution +// since we don't read back Lp-Trickle-Seq from the POST response. +func (c *TricklePublisher) resolveNextSeq() int { + url := fmt.Sprintf("%s/next", c.baseURL) + resp, err := c.client.Get(url) + if err != nil { + slog.Debug("Trickle /next probe failed; assuming fresh channel", "url", url, "err", err) + return 0 + } + defer resp.Body.Close() + + latest := resp.Header.Get("Lp-Trickle-Latest") + if latest == "" { + slog.Debug("Trickle /next missing Lp-Trickle-Latest; assuming fresh channel", "url", url, "status", resp.StatusCode) + return 0 + } + seq, err := strconv.Atoi(latest) + if err != nil { + slog.Debug("Trickle /next returned invalid Lp-Trickle-Latest; assuming fresh channel", "url", url, "value", latest, "err", err) + return 0 + } + slog.Debug("Trickle resolved seq from /next", "url", url, "seq", seq) + return seq +} + // NB expects to have the lock already since we mutate the index func (c *TricklePublisher) preconnect() (*pendingPost, error) { diff --git a/trickle/trickle_test.go b/trickle/trickle_test.go index f5cd178904..cad7fb7c1c 100644 --- a/trickle/trickle_test.go +++ b/trickle/trickle_test.go @@ -413,3 +413,84 @@ func makeServerWithServer(t *testing.T) (*require.Assertions, string, *Server) { func subConfig(t *testing.T, url string) TrickleSubscriberConfig { return TrickleSubscriberConfig{URL: url, Ctx: t.Context()} } + +// trickleProbeMux returns a mux that records POST slot URLs and serves a +// caller-supplied /next handler. Always handles DELETE on baseURL so the +// publisher's Close() doesn't error out. +func trickleProbeMux(channel string, nextHandler http.HandlerFunc) (*http.ServeMux, <-chan string) { + postedIdx := make(chan string, 8) + mux := http.NewServeMux() + mux.HandleFunc("GET "+channel+"/next", nextHandler) + mux.HandleFunc("POST "+channel+"/{idx}", func(w http.ResponseWriter, r *http.Request) { + // Drain the body first so we record completion order (which is + // deterministic — the segment with actual data closes its pipe + // first, the preconnected next segment closes only on Close()). + // Recording on invocation would race with goroutine scheduling. + io.Copy(io.Discard, r.Body) + select { + case postedIdx <- r.PathValue("idx"): + default: + } + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("DELETE "+channel, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + return mux, postedIdx +} + +// TestTrickle_PublisherProbeFallback verifies NewTricklePublisher gracefully +// falls back to slot 0 on a server that doesn't expose /next. Existing +// tests already exercise this path implicitly on master (no /next route), +// but pinning it explicitly documents the contract: probe failure MUST +// resolve to a fresh-channel default, never to a sentinel that races the +// server's own slot resolution. +func TestTrickle_PublisherProbeFallback(t *testing.T) { + require := require.New(t) + + mux, postedIdx := trickleProbeMux("/testest", func(w http.ResponseWriter, r *http.Request) { + // Simulate a pre-#3925 orchestrator: no /next route → 400 with no header. + http.Error(w, "Invalid idx", http.StatusBadRequest) + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + pub, err := NewTricklePublisher(ts.URL + "/testest") + require.Nil(err) + require.Nil(pub.Write(bytes.NewReader([]byte("first")))) + require.Nil(pub.Close()) + + select { + case idx := <-postedIdx: + require.Equal("0", idx, "first POST should target slot 0 on probe failure") + case <-time.After(2 * time.Second): + t.Fatal("first POST never arrived at server") + } +} + +// TestTrickle_PublisherProbeSuccess verifies NewTricklePublisher reads the +// server-reported nextWrite from /next and posts the first segment there. +// Uses a mock /next that returns 7 to confirm the publisher honors the +// probed value rather than always starting at 0. +func TestTrickle_PublisherProbeSuccess(t *testing.T) { + require := require.New(t) + + mux, postedIdx := trickleProbeMux("/testest", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Lp-Trickle-Latest", "7") + w.Write([]byte("7")) + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + pub, err := NewTricklePublisher(ts.URL + "/testest") + require.Nil(err) + require.Nil(pub.Write(bytes.NewReader([]byte("first")))) + require.Nil(pub.Close()) + + select { + case idx := <-postedIdx: + require.Equal("7", idx, "first POST should target server-reported nextWrite") + case <-time.After(2 * time.Second): + t.Fatal("first POST never arrived at server") + } +}