- 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
305 lines
9.6 KiB
Markdown
305 lines
9.6 KiB
Markdown
# 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**
|
|
```go
|
|
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:
|
|
```go
|
|
// 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**
|
|
```go
|
|
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**
|
|
```bash
|
|
cd backend/playout && go get github.com/fsnotify/fsnotify
|
|
```
|
|
|
|
**Step 2: Implement segment publish goroutine**
|
|
```go
|
|
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**
|
|
```go
|
|
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**
|
|
```go
|
|
// 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**
|
|
```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**
|
|
```typescript
|
|
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`:
|
|
```typescript
|
|
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**
|
|
```go
|
|
var (
|
|
activeItem string
|
|
playlistRevision string
|
|
lastSegmentSeq uint64
|
|
lastR2PubTime time.Time
|
|
consecutiveFails int
|
|
)
|
|
```
|
|
|
|
**Step 2: Update HealthHandler**
|
|
```go
|
|
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
|
|
|
|
1. Task 1 → Task 2 → Task 3 (coupled: naming enables sync, sync enables atomic publish)
|
|
2. Task 4 (quick constant change, can be batched with Task 3)
|
|
3. Task 5 (independent frontend work)
|
|
4. 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 |
|