- Added run ID + unique segment naming (r-{ISOTIME}/seg-{seq}.m4s)
- Added fsnotify watcher for real-time R2 segment upload (~7s)
- Added atomic manifest publication (temp key → CopyObject)
- Fixed manifest Cache-Control: public, max-age=1, s-maxage=2, must-revalidate
- Added hls.js error recovery: bounded retries (5x), exponential backoff, reconnecting UI
- Enriched /health endpoint: run_id, active_item, last_segment_seq, last_r2_pub, consecutive_failures, unhealthy detection
- Updated tests: HealthHandler (new fields + unhealthy), player error recovery test
- Updated TODO.md, BACKLOG.md, DONE.md, MEMORY.md
9.6 KiB
Production Readiness — Streaming Pipeline Gaps
For Hermes: Use subagent-driven-development skill to implement this plan task-by-task.
Goal: Close all HIGH/MEDIUM streaming pipeline gaps from STREAMING.md contract to reach production-readiness for linearvoid.tv.
Architecture: Modify backend/playout/src/playlist.go (Go playout engine) for R2 sync, segment naming, and health endpoint. Modify frontend/src/main.ts (TypeScript SPA) for player error recovery. Infrastructure-only (CORS on R2) is documented but needs manual Cloudflare action.
Tech Stack: Go (playout), TypeScript (frontend SPA), FFmpeg, AWS SDK for R2, hls.js
Priority 1: Continuous R2 Sync + Unique Segment Naming 🔴
These are tightly coupled — unique naming enables safe per-segment upload.
Task 1: Add run/session ID and monotonic media sequence
Objective: Generate a unique run ID on playout start and a monotonically increasing segment counter used in segment filenames.
Files:
- Modify:
backend/playout/src/playlist.go(global vars, FFmpeg args, syncLatestToR2) - Modify:
backend/playout/src/playlist_test.go(update tests for new naming)
Step 1: Add run ID and sequence to global state
var (
runID string // e.g. "r-20260810T170412Z"
segSequence uint64 // monotonically increasing segment counter
)
func initRunID() {
runID = "r-" + time.Now().UTC().Format("20060102T150405Z")
}
Step 2: Update FFmpeg segment filename pattern Replace strftime pattern with run ID + sequence:
// Current (bad):
"-hls_segment_filename", "/app/hls/segment_%Y%m%d_%H%M%S.m4s",
// New (unique):
"-hls_segment_filename", fmt.Sprintf("/app/hls/%s/seg-%%06d.m4s", runID),
Step 3: Update syncLatestToR2 for new naming
func syncLatestToR2(hlsDir string) error {
runDir := filepath.Join(hlsDir, runID)
files, err := filepath.Glob(filepath.Join(runDir, "seg-*.m4s"))
// Upload to R2: live/{runID}/seg-{seq}.m4s
}
Verification:
- Run:
cd backend/playout && go test -v -run TestLoadPlaylist - Expected: Playlist tests still pass with new naming
- Manual check: inspect hls directory for
r-*subdirectory structure
Task 2: Add filesystem watch for per-segment R2 upload
Objective: Upload each segment to R2 as FFmpeg completes it, instead of waiting for FFmpeg to exit. Use a filesystem watcher (fsnotify) triggered by new .m4s files.
Files:
- Modify:
backend/playout/src/playlist.go(add watcher goroutine) - Modify:
backend/playout/go.mod(add fsnotify dependency)
Step 1: Add fsnotify dependency
cd backend/playout && go get github.com/fsnotify/fsnotify
Step 2: Implement segment publish goroutine
func watchAndPublishSegments(ctx context.Context, hlsDir string) {
watcher, _ := fsnotify.NewWatcher()
defer watcher.Close()
watcher.Add(hlsDir)
for {
select {
case event := <-watcher.Events:
if event.Op&fsnotify.Create != 0 && strings.HasSuffix(event.Name, ".m4s") {
// Wait for file to be fully written
time.Sleep(500 * time.Millisecond)
uploadSegment(event.Name)
}
case <-ctx.Done():
return
}
}
}
Step 3: Wire watcher into main() loop Start the watcher goroutine before FFmpeg command, cancel context on exit.
Verification:
- Run:
go test -v ./...in playout directory - Manual: start playout, check R2 bucket for segments appearing in real-time
- Expected: new segments appear in R2 within ~7s of creation
Task 3: Atomic manifest publication
Objective: Upload manifest to temp key (live/playlist.m3u8.tmp), then atomically copy/replace to live/playlist.m3u8.
Files:
- Modify:
backend/playout/src/playlist.go(syncLatestToR2 / publishManifest)
Step 1: Write publishManifest function
func publishManifest(localPath string) error {
// 1. Upload to temp key
if err := SyncToR2(localPath, "live/playlist.m3u8.tmp"); err != nil {
return err
}
// 2. Atomic copy to final key (R2 supports CopyObject)
_, err := r2Client.CopyObject(context.TODO(), &s3.CopyObjectInput{
Bucket: aws.String(bucket),
CopySource: aws.String(bucket + "/live/playlist.m3u8.tmp"),
Key: aws.String("live/playlist.m3u8"),
})
return err
}
Verification:
- Run:
go test -v -run TestSyncToR2_ManifestCacheHeaders - Expected: manifest appears atomically in R2
- Manual: curl
media.linearvoid.tv/hls/playlist.m3u8— should never 404
Task 4: Fix manifest Cache-Control contract
Objective: Align manifest cache headers with STREAMING.md contract.
Files:
- Modify:
backend/playout/src/playlist.go(constants)
Step 1: Update constant
// Current (wrong):
manifestCacheControl = "public, max-age=2"
// New (per contract):
manifestCacheControl = "public, max-age=1, s-maxage=2, must-revalidate"
Verification:
- Run:
go test -v -run TestSyncToR2_ManifestCacheHeaders - Expected: test still passes with new header values
Priority 2: Player Error Recovery 🟡
Task 5: Add hls.js error handler with retry logic
Objective: Handle hls.js errors with bounded retries, exponential backoff, and a visible reconnecting state.
Files:
- Modify:
frontend/src/main.ts(initPlayerAndChat, add error handler) - Modify:
frontend/src/style.css(reconnecting overlay style) - Modify:
frontend/index.html(add reconnecting indicator element)
Step 1: Add reconnecting overlay HTML
<div id="reconnecting-overlay" class="reconnecting-overlay hidden">
<div class="reconnecting-spinner"></div>
<span>Reconnecting...</span>
</div>
Step 2: Implement error handler in main.ts
let errorCount = 0;
const MAX_RETRIES = 5;
const BASE_DELAY_MS = 2000;
hls.on(Hls.Events.ERROR, (_event, data) => {
if (data.fatal) {
switch (data.type) {
case Hls.ErrorTypes.NETWORK_ERROR:
errorCount++;
if (errorCount <= MAX_RETRIES) {
const delay = BASE_DELAY_MS * Math.pow(2, errorCount - 1);
showReconnecting();
setTimeout(() => hls?.startLoad(), delay);
} else {
showReconnecting('Stream unavailable');
}
break;
case Hls.ErrorTypes.MEDIA_ERROR:
errorCount++;
if (errorCount <= 1) {
hls?.recoverMediaError();
} else {
hls?.destroy();
showReconnecting('Stream unavailable');
}
break;
default:
hls?.destroy();
showReconnecting('Stream unavailable');
break;
}
}
});
// Add drift detection
hls.on(Hls.Events.LEVEL_UPDATED, () => {
const latency = hls.latency;
if (latency > 35) {
hls.currentLevel = -1; // seek to live edge
}
});
Step 3: Write test for error recovery
Add to frontend/src/main.test.ts:
it("retries on network error with exponential backoff", async () => {
// Mock hls.js error event, verify retry with backoff
});
Verification:
- Run:
cd frontend && npx vitest run - Expected: 31 existing tests pass + new error recovery test
- Manual: kill playout process, verify player shows "Reconnecting..." and recovers when playout restarts
Priority 3: Health Endpoint Detail 🟢
Task 6: Enrich /health endpoint
Objective: Add active item, playlist revision, run ID, last segment info, last R2 pub timestamp, consecutive failures to /health.
Files:
- Modify:
backend/playout/src/playlist.go(HealthHandler, global state)
Step 1: Add tracking state
var (
activeItem string
playlistRevision string
lastSegmentSeq uint64
lastR2PubTime time.Time
consecutiveFails int
)
Step 2: Update HealthHandler
func HealthHandler(w http.ResponseWriter, r *http.Request) {
resp := map[string]any{
"status": status,
"service": "playout",
"mode": playoutMode,
"restarts": restartCount,
"run_id": runID,
"active_item": activeItem,
"playlist_revision": playlistRevision,
"last_segment_seq": lastSegmentSeq,
"last_r2_pub": lastR2PubTime.UTC().Format(time.RFC3339),
"consecutive_failures": consecutiveFails,
"timestamp": time.Now().UTC().Format(time.RFC3339),
}
json.NewEncoder(w).Encode(resp)
}
Verification:
- Run:
go test -v -run TestHealthHandler - Manual:
curl localhost:8081/health | jq - Expected: all new fields present and populated
Execution Order
- Task 1 → Task 2 → Task 3 (coupled: naming enables sync, sync enables atomic publish)
- Task 4 (quick constant change, can be batched with Task 3)
- Task 5 (independent frontend work)
- Task 6 (quick, can be batched with Task 1-3)
Total estimated changes: ~200 lines Go, ~60 lines TypeScript, ~20 lines HTML, ~10 lines CSS
Files Changed
| File | Purpose |
|---|---|
backend/playout/src/playlist.go |
Core changes: run ID, unique naming, filesystem watcher, atomic publish, health |
backend/playout/src/playlist_test.go |
Update for new naming + add watcher tests |
backend/playout/go.mod |
Add fsnotify |
frontend/src/main.ts |
Player error recovery |
frontend/src/style.css |
Reconnecting overlay styles |
frontend/index.html |
Reconnecting overlay element |
frontend/src/main.test.ts |
Error recovery tests |