trickle: align Go publisher with SDK probe-and-fallback pattern

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) {
 

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