Update trickle local publisher streaming behavior

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)

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
 		}

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

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