From 1224f8786701e15e2b7bfd6a2456a447786c8006 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Wed, 15 Jul 2026 15:03:47 -0700 Subject: [PATCH 1/2] trickle: advance local publisher stream head --- trickle/README.md | 8 + trickle/local_publisher.go | 12 +- trickle/segment_buffer.go | 146 ++++++++++++ trickle/segment_buffer_test.go | 68 ++++++ trickle/trickle_server.go | 127 ++++++++-- trickle/trickle_test.go | 422 ++++++++++++++++++++++++++++++++- 6 files changed, 758 insertions(+), 25 deletions(-) create mode 100644 trickle/segment_buffer.go create mode 100644 trickle/segment_buffer_test.go diff --git a/trickle/README.md b/trickle/README.md index ce4a8706b5..bd2cfd69e3 100644 --- a/trickle/README.md +++ b/trickle/README.md @@ -50,6 +50,11 @@ Servers may offer some grace with leading sequence numbers to avoid data races, Publishers are responsible for segmenting content (if necessary) and subscribers are responsible for re-assembling content (if necessary) +Successful publisher POST responses include `Lp-Trickle-Seq` metadata (HTTP header) with the effective segment index written by the server. + +If a publisher sends `Lp-Trickle-Reset`, the server treats it as a restart signal for any `seq` value. +The server closes prior segments to unblock waiting subscribers while still allowing preconnected/ahead segments. + Subscribers can initiate a subscribe with a `seq` of -1 to retrieve the most recent publish. With preconnects, the subscriber may be waiting for the *next* publish. For video this allows clients to eg, start streaming at the live edge of the next GOP. Subscribers can retrieve the current `seq` with the `Lp-Trickle-Seq` metadata (HTTP header). This is useful in case `-1` was used to initiate the subscription; the subscribing client can then pre-connect to `Lp-Trickle-Seq + 1` @@ -58,6 +63,9 @@ GET `/channel-name/next` returns the next segment seq as plain text in the respo If the channel does not exist, the server returns 404. If the channel is closed, the server also includes `Lp-Trickle-Closed: terminated`. +`Lp-Trickle-Seq` from a publisher write using `seq=-1` is not sufficient by itself to drive real-time ordering if writes overlap. +It is still useful for post-facto mapping/observability, reconciliation after segment close, and debugging dropped/misordered assumptions. + Subscribers can initiate a subscribe with a `seq` of -N to get the Nth-from-last segment. (TODO) The server should send subscribers `Lp-Trickle-Size` metadata to indicate the size of the content up until now. This allows clients to know where the live edge is, eg video implementations can decode-and-discard frames up until the edge to achieve immediate playback without waiting for the next segment. (TODO) diff --git a/trickle/local_publisher.go b/trickle/local_publisher.go index bc0082dd8e..9fe9aae481 100644 --- a/trickle/local_publisher.go +++ b/trickle/local_publisher.go @@ -5,6 +5,7 @@ import ( "io" "log/slog" "sync" + "time" ) // local (in-memory) publisher for trickle protocol @@ -32,7 +33,10 @@ func (c *TrickleLocalPublisher) CreateChannel() { } func (c *TrickleLocalPublisher) Write(data io.Reader) error { - stream := c.server.getOrCreateStream(c.channelName, c.mimeType, true) + stream := c.server.getOrCreateStream(c.channelName, c.mimeType, false) + if stream == nil { + return StreamNotFoundErr + } c.mu.Lock() seq := c.seq segment, exists := stream.getForWrite(seq) @@ -56,6 +60,12 @@ func (c *TrickleLocalPublisher) Write(data io.Reader) error { for { n, err := data.Read(buf) if n > 0 { + if totalRead == 0 { + stream.mutex.Lock() + stream.nextWrite = seq + 1 + stream.writeTime = time.Now() + stream.mutex.Unlock() + } segment.writeData(buf[:n]) totalRead += n } diff --git a/trickle/segment_buffer.go b/trickle/segment_buffer.go new file mode 100644 index 0000000000..896b1656b2 --- /dev/null +++ b/trickle/segment_buffer.go @@ -0,0 +1,146 @@ +package trickle + +import ( + "sync/atomic" +) + +const ( + segmentBufferInitialPageSize = 32 * 1024 + segmentBufferMaxPageSize = 1024 * 1024 +) + +type segmentPage struct { + start int + data []byte + written int +} + +type segmentPageList struct { + pages []*segmentPage +} + +// segmentBuffer is an append-only paged buffer tailored for Segment fanout. +// Pages are immutable containers whose published prefix is tracked atomically, +// allowing readers to take a lock-free fast path when data is already available. +type segmentBuffer struct { + pageList atomic.Pointer[segmentPageList] + published atomic.Int64 + totalWritten int + initialCap int + maxPageCap int + nextPageCap int +} + +func newSegmentBuffer() *segmentBuffer { + return newSegmentBufferWithPageCaps(segmentBufferInitialPageSize, segmentBufferMaxPageSize) +} + +func newSegmentBufferWithPageCaps(initialCap, maxPageCap int) *segmentBuffer { + if initialCap <= 0 { + initialCap = segmentBufferInitialPageSize + } + if maxPageCap < initialCap { + maxPageCap = initialCap + } + b := &segmentBuffer{ + initialCap: initialCap, + maxPageCap: maxPageCap, + nextPageCap: initialCap, + } + b.pageList.Store(&segmentPageList{pages: nil}) + return b +} + +func (b *segmentBuffer) write(data []byte) { + for len(data) > 0 { + page := b.ensureTailPage() + written := page.written + remaining := cap(page.data) - written + if remaining > len(data) { + remaining = len(data) + } + + end := written + remaining + copy(page.data[written:end], data[:remaining]) + page.written = end + + data = data[remaining:] + b.totalWritten += remaining + } + b.published.Store(int64(b.totalWritten)) +} + +func (b *segmentBuffer) ensureTailPage() *segmentPage { + list := b.pageList.Load() + if list != nil && len(list.pages) > 0 { + tail := list.pages[len(list.pages)-1] + if tail.written < cap(tail.data) { + return tail + } + } + + pageCap := b.nextPageCap + if pageCap == 0 { + pageCap = b.initialCap + } + page := &segmentPage{ + start: b.totalWritten, + data: make([]byte, pageCap), + } + newPages := make([]*segmentPage, 0, len(list.pages)+1) + newPages = append(newPages, list.pages...) + newPages = append(newPages, page) + b.pageList.Store(&segmentPageList{pages: newPages}) + + if b.nextPageCap < b.maxPageCap { + b.nextPageCap *= 2 + if b.nextPageCap > b.maxPageCap { + b.nextPageCap = b.maxPageCap + } + } + return page +} + +func (b *segmentBuffer) isEmpty() bool { + return b.published.Load() == 0 +} + +func (b *segmentBuffer) readChunk(pos int) ([]byte, int, bool, bool, int) { + if pos < 0 { + return nil, pos, false, true, int(b.published.Load()) + } + published := int(b.published.Load()) + if pos > published { + return nil, pos, false, true, published + } + if pos == published { + return nil, pos, false, false, published + } + + list := b.pageList.Load() + if list == nil { + return nil, pos, false, false, published + } + + for i, page := range list.pages { + pageEnd := published + if i+1 < len(list.pages) { + pageEnd = list.pages[i+1].start + if pageEnd > published { + pageEnd = published + } + } + if pos < page.start || pos >= pageEnd { + continue + } + + offset := pos - page.start + data := page.data[offset : pageEnd-page.start] + return data, pageEnd, pageEnd == published, false, published + } + + // We may have observed a newer published cursor than the currently visible + // page-list snapshot. Treat this as "not available yet" so callers can retry + // under synchronization instead of incorrectly signaling EOF. + return nil, pos, false, false, published +} diff --git a/trickle/segment_buffer_test.go b/trickle/segment_buffer_test.go new file mode 100644 index 0000000000..b733611c47 --- /dev/null +++ b/trickle/segment_buffer_test.go @@ -0,0 +1,68 @@ +package trickle + +import ( + "bytes" + "testing" +) + +func TestSegmentBufferReadChunkAcrossPages(t *testing.T) { + buf := newSegmentBuffer() + chunkA := bytes.Repeat([]byte("a"), segmentBufferInitialPageSize) + chunkB := bytes.Repeat([]byte("b"), segmentBufferInitialPageSize*2) + want := append(append([]byte{}, chunkA...), chunkB...) + + buf.write(chunkA) + buf.write(chunkB) + + var got []byte + nextPos := 0 + for { + data, next, atTail, invalid, _ := buf.readChunk(nextPos) + if invalid { + t.Fatalf("invalid cursor at pos=%d", nextPos) + } + if len(data) == 0 { + break + } + got = append(got, data...) + nextPos = next + if atTail { + break + } + } + if !bytes.Equal(got, want) { + t.Fatalf("buffer mismatch: got=%d want=%d", len(got), len(want)) + } + if nextPos != len(want) { + t.Fatalf("unexpected tail cursor: pos=%d", nextPos) + } +} + +func TestSegmentBufferTailGrowth(t *testing.T) { + buf := newSegmentBuffer() + chunkA := []byte("hello") + chunkB := []byte(" world") + + buf.write(chunkA) + + data, nextPos, atTail, invalid, _ := buf.readChunk(0) + if len(data) == 0 || !atTail || invalid { + t.Fatalf("initial read state len=%d atTail=%v invalid=%v", len(data), atTail, invalid) + } + if string(data) != "hello" { + t.Fatalf("unexpected initial data %q", string(data)) + } + + buf.write(chunkB) + + data, nextPos, atTail, invalid, _ = buf.readChunk(nextPos) + if len(data) == 0 || !atTail || invalid { + t.Fatalf("growth read state len=%d atTail=%v invalid=%v", len(data), atTail, invalid) + } + if string(data) != " world" { + t.Fatalf("unexpected growth data %q", string(data)) + } + if nextPos != len(chunkA)+len(chunkB) { + t.Fatalf("unexpected grown cursor: pos=%d", nextPos) + } +} diff --git a/trickle/trickle_server.go b/trickle/trickle_server.go index 895677f9cf..b08a3a858c 100644 --- a/trickle/trickle_server.go +++ b/trickle/trickle_server.go @@ -13,6 +13,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" ) @@ -36,6 +37,14 @@ type TrickleServerConfig struct { // How often to sweep for idle channels (default 1 minute) SweepInterval time.Duration + + // BeforeCreate runs before HTTP channel creation. + // Return RequestError for expected client/policy failures. + BeforeCreate func(r *http.Request, streamName string) error + + // BeforeDelete runs before HTTP channel deletion. + // Return RequestError for expected client/policy failures. + BeforeDelete func(r *http.Request, streamName string) error } type Server struct { @@ -62,8 +71,9 @@ type Segment struct { idx int mutex *sync.Mutex cond *sync.Cond - buffer *bytes.Buffer + buffer *segmentBuffer closed bool + done atomic.Bool // to shut down any pending publishers closeCh chan bool @@ -83,6 +93,29 @@ const maxSegmentsPerStream = 5 var FirstByteTimeout = errors.New("pending read timeout") +// RequestError represents an expected request or policy failure from a hook. +type RequestError struct { + StatusCode int + Message string +} + +func (e *RequestError) Error() string { + if e.Message != "" { + return e.Message + } + if e.StatusCode != 0 { + return http.StatusText(e.StatusCode) + } + return http.StatusText(http.StatusBadRequest) +} + +func (e *RequestError) httpStatus() int { + if e.StatusCode >= 400 && e.StatusCode < 500 { + return e.StatusCode + } + return http.StatusBadRequest +} + func applyDefaults(config *TrickleServerConfig) { if config.BasePath == "" { config.BasePath = "/" @@ -259,6 +292,12 @@ func (sm *Server) closeStream(streamName string) error { func (sm *Server) handleDelete(w http.ResponseWriter, r *http.Request) { streamName := r.PathValue("streamName") + if sm.config.BeforeDelete != nil { + if err := sm.config.BeforeDelete(r, streamName); err != nil { + writeHookError(w, err) + return + } + } if err := sm.closeStream(streamName); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -288,13 +327,30 @@ func (sm *Server) closeSeq(w http.ResponseWriter, r *http.Request) { } func (sm *Server) handleCreate(w http.ResponseWriter, r *http.Request) { - stream := sm.getOrCreateStream(r.PathValue("streamName"), r.Header.Get("Expect-Content"), false) + streamName := r.PathValue("streamName") + mimeType := r.Header.Get("Expect-Content") + if sm.config.BeforeCreate != nil { + if err := sm.config.BeforeCreate(r, streamName); err != nil { + writeHookError(w, err) + return + } + } + stream := sm.getOrCreateStream(streamName, mimeType, false) if stream == nil { http.Error(w, "Stream not found", http.StatusNotFound) return } } +func writeHookError(w http.ResponseWriter, err error) { + var requestErr *RequestError + if errors.As(err, &requestErr) { + http.Error(w, requestErr.Error(), requestErr.httpStatus()) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) +} + func (sm *Server) handlePost(w http.ResponseWriter, r *http.Request) { stream := sm.getOrCreateStream(r.PathValue("streamName"), r.Header.Get("Content-Type"), false) if stream == nil { @@ -374,6 +430,19 @@ func (tr *timeoutReader) Close() error { func (s *Stream) handlePost(w http.ResponseWriter, r *http.Request, idx int) { segment, _ := s.getForWrite(idx) + if r.Header.Get("Lp-Trickle-Reset") != "" { + // Usually means the publisher had to restart for some reason. + // Close prior segments to unblock subscribers for any hanging writes + // but allow for preconnected segments (sometimes they come out-of-order) + s.mutex.Lock() + for _, seg := range s.segments { + if seg != nil && seg.idx < segment.idx { + seg.close() + } + } + s.mutex.Unlock() + } + // Wrap the request body with the custom timeoutReader so we can send // provisional headers (keepalives) until receiving the first byte reader := &timeoutReader{ @@ -394,7 +463,7 @@ func (s *Stream) handlePost(w http.ResponseWriter, r *http.Request, idx int) { if totalRead == 0 { startedAt = time.Now() s.mutex.Lock() - s.nextWrite = idx + 1 + s.nextWrite = segment.idx + 1 s.writeTime = startedAt s.mutex.Unlock() } @@ -418,9 +487,10 @@ func (s *Stream) handlePost(w http.ResponseWriter, r *http.Request, idx int) { s.mutex.Lock() isClosed := s.closed // increment seq anyway: avoids clients erroring out on next seq - s.nextWrite = idx + 1 + s.nextWrite = segment.idx + 1 s.writeTime = startedAt s.mutex.Unlock() + w.Header().Set("Lp-Trickle-Seq", strconv.Itoa(segment.idx)) if isClosed { w.Header().Set("Lp-Trickle-Closed", "terminated") } @@ -440,6 +510,7 @@ func (s *Stream) handlePost(w http.ResponseWriter, r *http.Request, idx int) { } // Mark segment as closed + w.Header().Set("Lp-Trickle-Seq", strconv.Itoa(segment.idx)) segment.close() slog.Info("POST completed", "stream", s.name, "idx", idx, "bytes", totalRead, "took", time.Since(startedAt)) } @@ -588,14 +659,9 @@ func (s *Stream) handleGet(w http.ResponseWriter, r *http.Request, idx int) { latestSeq := s.nextWrite s.mutex.RUnlock() w.Header().Set("Lp-Trickle-Seq", strconv.Itoa(segment.idx)) + w.Header().Set("Lp-Trickle-Latest", strconv.Itoa(latestSeq)) if closed { w.Header().Set("Lp-Trickle-Closed", "terminated") - } else { - // usually happens if a publisher cancels a pending segment right before closing the channel - // other times, the subscriber is slow and the segment falls out of the live window - // send over latest seq so slow clients can grab leading edge - w.Header().Set("Lp-Trickle-Latest", strconv.Itoa(latestSeq)) - w.WriteHeader(470) } } return totalWrites, nil @@ -614,7 +680,7 @@ func newSegment(idx int) *Segment { mu := &sync.Mutex{} return &Segment{ idx: idx, - buffer: new(bytes.Buffer), + buffer: newSegmentBuffer(), cond: sync.NewCond(mu), mutex: mu, closeCh: make(chan bool), @@ -626,29 +692,43 @@ func (segment *Segment) writeData(data []byte) { defer segment.mutex.Unlock() // Write to buffer - segment.buffer.Write(data) + segment.buffer.write(data) // Signal waiting readers segment.cond.Broadcast() } -func (s *Segment) readData(startPos int) ([]byte, bool) { +func (s *Segment) readData(readPos int) ([]byte, int, bool) { + data, nextPos, atTail, invalid, _ := s.buffer.readChunk(readPos) + if invalid { + slog.Info("Invalid start pos, invoking eof") + return nil, readPos, true + } + if len(data) > 0 { + // A concurrent write+close can land after readChunk() snapshots published. + // Only signal EOF from the fast path if a fresh published load still + // matches the returned cursor. + if atTail && s.done.Load() && nextPos == int(s.buffer.published.Load()) { + return data, nextPos, true + } + return data, nextPos, false + } + s.mutex.Lock() defer s.mutex.Unlock() for { - totalLen := s.buffer.Len() - if startPos < totalLen { - data := s.buffer.Bytes()[startPos:totalLen] - return data, s.closed + data, nextPos, atTail, invalid, published := s.buffer.readChunk(readPos) + if len(data) > 0 { + return data, nextPos, s.closed && atTail && nextPos == published } - if startPos > totalLen { + if invalid { slog.Info("Invalid start pos, invoking eof") // This might happen if the buffer was reset // eg because of a repeated POST - return nil, true + return nil, readPos, true } if s.closed { - return nil, true + return nil, readPos, true } // Wait for new data s.cond.Wait() @@ -663,6 +743,7 @@ func (s *Segment) close() { defer s.mutex.Unlock() if !s.closed { s.closed = true + s.done.Store(true) close(s.closeCh) s.cond.Broadcast() } @@ -672,11 +753,11 @@ func (s *Segment) isFresh() bool { // fresh segments have not been written to yet s.mutex.Lock() defer s.mutex.Unlock() - return !s.closed && s.buffer.Len() == 0 + return !s.closed && s.buffer.isEmpty() } func (ss *SegmentSubscriber) readData() ([]byte, bool) { - data, eof := ss.segment.readData(ss.readPos) - ss.readPos += len(data) + data, nextPos, eof := ss.segment.readData(ss.readPos) + ss.readPos = nextPos return data, eof } diff --git a/trickle/trickle_test.go b/trickle/trickle_test.go index f5cd178904..bac63d9b9f 100644 --- a/trickle/trickle_test.go +++ b/trickle/trickle_test.go @@ -97,6 +97,319 @@ func TestTrickle_Close(t *testing.T) { require.Error(StreamNotFoundErr, pub2.Write(bytes.NewReader([]byte("bad post")))) } +func TestLocalPublisher_CreateContract(t *testing.T) { + t.Run("write without autocreate returns stream not found", func(t *testing.T) { + require := require.New(t) + server := ConfigureServer(TrickleServerConfig{ + Mux: http.NewServeMux(), // unused in practice with local-only publishing + Autocreate: false, + }) + pub := NewLocalPublisher(server, "missing", "text/plain") + + err := pub.Write(bytes.NewReader([]byte("hello"))) + + require.ErrorIs(err, StreamNotFoundErr) + _, exists := server.getStream("missing") + require.False(exists) + }) + + t.Run("create channel without autocreate then write succeeds", func(t *testing.T) { + require := require.New(t) + server := ConfigureServer(TrickleServerConfig{ + Mux: http.NewServeMux(), // unused in practice with local-only publishing + Autocreate: false, + }) + pub := NewLocalPublisher(server, "created", "text/plain") + + pub.CreateChannel() + err := pub.Write(bytes.NewReader([]byte("hello"))) + + require.NoError(err) + _, exists := server.getStream("created") + require.True(exists) + }) + + t.Run("write with autocreate creates missing channel", func(t *testing.T) { + require := require.New(t) + server := ConfigureServer(TrickleServerConfig{ + Mux: http.NewServeMux(), // unused in practice with local-only publishing + Autocreate: true, + }) + pub := NewLocalPublisher(server, "autocreated", "text/plain") + + err := pub.Write(bytes.NewReader([]byte("hello"))) + + require.NoError(err) + _, exists := server.getStream("autocreated") + require.True(exists) + }) +} + +func TestTrickle_HTTPCreateContract(t *testing.T) { + t.Run("create without autocreate returns not found", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: false, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/missing", "text/plain", nil) + require.NoError(err) + resp.Body.Close() + + require.Equal(http.StatusNotFound, resp.StatusCode) + }) + + t.Run("publish without autocreate returns not found", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: false, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/missing/0", "text/plain", bytes.NewReader([]byte("hello"))) + require.NoError(err) + resp.Body.Close() + + require.Equal(http.StatusNotFound, resp.StatusCode) + }) + + t.Run("create with autocreate succeeds", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: true, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/created", "text/plain", nil) + require.NoError(err) + resp.Body.Close() + + require.Equal(http.StatusOK, resp.StatusCode) + }) + + t.Run("publish with autocreate succeeds", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: true, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/published/0", "text/plain", bytes.NewReader([]byte("hello"))) + require.NoError(err) + resp.Body.Close() + + require.Equal(http.StatusOK, resp.StatusCode) + }) +} + +func TestTrickle_BeforeCreate(t *testing.T) { + t.Run("called before creation", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + var called bool + var server *Server + server = ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: true, + BeforeCreate: func(r *http.Request, streamName string) error { + called = true + require.Equal("created", streamName) + require.Equal("text/plain", r.Header.Get("Expect-Content")) + _, exists := server.getStream(streamName) + require.False(exists) + return nil + }, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + req, err := http.NewRequest(http.MethodPost, ts.URL+"/created", nil) + require.NoError(err) + req.Header.Set("Expect-Content", "text/plain") + resp, err := http.DefaultClient.Do(req) + require.NoError(err) + resp.Body.Close() + + require.Equal(http.StatusOK, resp.StatusCode) + require.True(called) + resp, err = http.Get(ts.URL + "/created/next") + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusOK, resp.StatusCode) + }) + + t.Run("request error prevents creation", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: true, + BeforeCreate: func(r *http.Request, streamName string) error { + return &RequestError{StatusCode: http.StatusForbidden, Message: "nope"} + }, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/blocked", "text/plain", nil) + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusForbidden, resp.StatusCode) + + resp, err = http.Get(ts.URL + "/blocked/next") + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusNotFound, resp.StatusCode) + }) + + t.Run("generic error prevents creation", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: true, + BeforeCreate: func(r *http.Request, streamName string) error { + return errors.New("boom") + }, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/errored", "text/plain", nil) + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusInternalServerError, resp.StatusCode) + + resp, err = http.Get(ts.URL + "/errored/next") + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusNotFound, resp.StatusCode) + }) +} + +func TestTrickle_BeforeDelete(t *testing.T) { + t.Run("called before deletion", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + var called bool + var server *Server + server = ConfigureServer(TrickleServerConfig{ + Mux: mux, + BeforeDelete: func(r *http.Request, streamName string) error { + called = true + require.Equal("deleted", streamName) + _, exists := server.getStream(streamName) + require.True(exists) + return nil + }, + }) + NewLocalPublisher(server, "deleted", "text/plain").CreateChannel() + ts := httptest.NewServer(mux) + defer ts.Close() + + req, err := http.NewRequest(http.MethodDelete, ts.URL+"/deleted", nil) + require.NoError(err) + resp, err := http.DefaultClient.Do(req) + require.NoError(err) + resp.Body.Close() + + require.Equal(http.StatusOK, resp.StatusCode) + require.True(called) + resp, err = http.Get(ts.URL + "/deleted/next") + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusNotFound, resp.StatusCode) + }) + + t.Run("request error prevents deletion", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + server := ConfigureServer(TrickleServerConfig{ + Mux: mux, + BeforeDelete: func(r *http.Request, streamName string) error { + return &RequestError{StatusCode: http.StatusUnauthorized, Message: "nope"} + }, + }) + NewLocalPublisher(server, "blocked-delete", "text/plain").CreateChannel() + ts := httptest.NewServer(mux) + defer ts.Close() + + req, err := http.NewRequest(http.MethodDelete, ts.URL+"/blocked-delete", nil) + require.NoError(err) + resp, err := http.DefaultClient.Do(req) + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusUnauthorized, resp.StatusCode) + + resp, err = http.Get(ts.URL + "/blocked-delete/next") + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusOK, resp.StatusCode) + }) + + t.Run("generic error prevents deletion", func(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + server := ConfigureServer(TrickleServerConfig{ + Mux: mux, + BeforeDelete: func(r *http.Request, streamName string) error { + return errors.New("boom") + }, + }) + NewLocalPublisher(server, "errored-delete", "text/plain").CreateChannel() + ts := httptest.NewServer(mux) + defer ts.Close() + + req, err := http.NewRequest(http.MethodDelete, ts.URL+"/errored-delete", nil) + require.NoError(err) + resp, err := http.DefaultClient.Do(req) + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusInternalServerError, resp.StatusCode) + + resp, err = http.Get(ts.URL + "/errored-delete/next") + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusOK, resp.StatusCode) + }) +} + +func TestTrickle_Delete(t *testing.T) { + require := require.New(t) + mux := http.NewServeMux() + ConfigureServer(TrickleServerConfig{ + Mux: mux, + Autocreate: true, + }) + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Post(ts.URL+"/delete-test", "text/plain", nil) + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusOK, resp.StatusCode) + + req, err := http.NewRequest(http.MethodDelete, ts.URL+"/delete-test", nil) + require.NoError(err) + resp, err = http.DefaultClient.Do(req) + require.NoError(err) + resp.Body.Close() + require.Equal(http.StatusOK, resp.StatusCode) +} + func TestTrickle_SetSeq(t *testing.T) { require, channelURL := makeServer(t) @@ -152,7 +465,7 @@ func TestTrickle_Reset(t *testing.T) { require.Nil(err) wg := &sync.WaitGroup{} - // give preconnects time to latch on and autocreate the channel + // give preconnects time to latch on time.Sleep(5 * time.Millisecond) respCh := make(chan *http.Response) @@ -228,6 +541,113 @@ func TestTrickle_Reset(t *testing.T) { wg.Wait() } +// TestTrickle_PublisherReset verifies reset behavior for publisher restarts: +// 1) a blocked subscriber on an open segment is unblocked by reset, +// 2) already-written bytes on that segment remain readable, +// 3) reset write goes to the current nextWrite index. +func TestTrickle_PublisherReset(t *testing.T) { + require, channelURL, server := makeServerWithServer(t) + + lp := NewLocalPublisher(server, "testest", "text/plain") + lp.CreateChannel() + + // Partial write of segment 0 via pipe - do not close pipe yet. + r0, w0 := io.Pipe() + writeDone := make(chan struct{}) + go func() { + defer close(writeDone) + _ = lp.Write(r0) + }() + _, err := w0.Write([]byte("Hello")) + require.Nil(err) + + // Local subscriber reads first bytes from segment 0, then blocks. + sub := NewLocalSubscriber(server, "testest") + sub.SetSeq(0) + td, err := sub.Read() + require.Nil(err) + + buf := make([]byte, 5) + _, err = io.ReadFull(td.Reader, buf) + require.Nil(err) + require.Equal("Hello", string(buf)) + + unblocked := make(chan []byte, 1) + go func() { + rest, _ := io.ReadAll(td.Reader) + unblocked <- rest + }() + + // HTTP POST reset - closes previous segments and writes next segment. + req, err := http.NewRequest("POST", channelURL+"/-1", bytes.NewReader([]byte("after-reset"))) + require.Nil(err) + req.Header.Set("Content-Type", "text/plain") + req.Header.Set("Lp-Trickle-Reset", "true") + resp, err := http.DefaultClient.Do(req) + require.Nil(err) + require.Equal(http.StatusOK, resp.StatusCode) + require.Equal("1", resp.Header.Get("Lp-Trickle-Seq"), "POST /-1 should echo resolved seq") + resp.Body.Close() + + // This receive deadlocks if reset doesn't unblock the reader. + <-unblocked + + // Re-read segment 0 - partial data should still be there. + sub.SetSeq(0) + td, err = sub.Read() + require.Nil(err) + data, err := io.ReadAll(td.Reader) + require.Nil(err) + require.Equal("Hello", string(data)) + + // Reset write should land at index 1. + td, err = sub.Read() + require.Nil(err) + data, err = io.ReadAll(td.Reader) + require.Nil(err) + require.Equal("after-reset", string(data)) + + // Additional writes to the old segment writer can still succeed until writer close. + // This is a bit racy w subscribers but subs will terminate once they catch up. + // NB: Fix this someday + _, err = w0.Write([]byte("late-bytes")) + require.NoError(err) + + require.Nil(w0.Close()) + <-writeDone +} + +func TestTrickle_EmptySegment(t *testing.T) { + require, channelURL := makeServer(t) + + pub, err := NewTricklePublisher(channelURL) + require.Nil(err) + defer pub.Close() + + pp, err := pub.Next() + require.Nil(err) + + n, err := pp.Write(bytes.NewReader(nil)) + require.Nil(err) + require.Equal(int64(0), n) + + sub, err := NewTrickleSubscriber(subConfig(t, channelURL)) + require.Nil(err) + sub.SetSeq(0) + + resp, err := sub.Read() + defer resp.Body.Close() + + require.Equal(http.StatusOK, resp.StatusCode) + require.Equal("0", resp.Header.Get("Lp-Trickle-Seq")) + require.Equal("", resp.Header.Get("Lp-Trickle-Closed")) + require.Equal("1", resp.Header.Get("Lp-Trickle-Latest")) + + body, err := io.ReadAll(resp.Body) + require.Nil(err) + require.Equal("", string(body)) +} + func TestTrickle_IdleSweep(t *testing.T) { require := require.New(t) mux := http.NewServeMux() From 7e1dbaf3285efd93e5dcfa10724209466c2c510e Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Thu, 16 Jul 2026 15:40:48 -0700 Subject: [PATCH 2/2] trickle: rename stream creation flag --- trickle/trickle_server.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/trickle/trickle_server.go b/trickle/trickle_server.go index b08a3a858c..23fcb95a7c 100644 --- a/trickle/trickle_server.go +++ b/trickle/trickle_server.go @@ -186,11 +186,11 @@ func (sm *Server) getStream(streamName string) (*Stream, bool) { return stream, exists } -func (sm *Server) getOrCreateStream(streamName, mimeType string, isLocal bool) *Stream { +func (sm *Server) getOrCreateStream(streamName, mimeType string, forceCreate bool) *Stream { sm.mutex.Lock() stream, exists := sm.streams[streamName] - if !exists && (isLocal || sm.config.Autocreate) { + if !exists && (forceCreate || sm.config.Autocreate) { stream = &Stream{ segments: make([]*Segment, maxSegmentsPerStream), name: streamName,