From bb8ffbbe4dc799870d12e256f9f13bafd0bf5ba9 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Mon, 6 Jul 2026 19:36:27 -0700 Subject: [PATCH] fix(livestream): don't let the idle-timeout finalizer end an active stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The idle-timeout finalizer (processFinalizeLivestreamTask) could set endedAt on the record an actively-streaming user was publishing under, taking the stream pre-live underneath a still-flowing ingest. Root cause: lastSeenAt only advances via the per-segment heartbeat (StreamSession.doUpdateLivestream), which is coupled to segment arrival and can lag behind actual ingestion. A ~60s ingest gap froze lastSeenAt; the 300s idle timer then fired while segments were still flowing, and endedAt was written onto the live record. Guard the finalizer: when lastSeenAt is stale but the record is still the streamer's latest livestream (no newer record supersedes it), reschedule the finalize for one more idle window instead of ending — giving a lagging heartbeat a chance to catch up. If the stream is truly abandoned, the heartbeat stays frozen and the next pass ends it. Superseded records (a newer one exists) are still ended, since they're no longer live. Also fix the related orphaned-finalize-task bug: startLivestream now ends any prior un-ended livestream for the repo before creating the new one, so the prior record's already-scheduled idle-timeout task hits its rec.EndedAt != nil early-skip instead of later writing a stale endedAt onto a replaced record. Verified against the 2026-07-06 production incident: segment manifests showed the stream went pre-live exactly when the idle timer fired on the live record (lastSeenAt frozen at the pre-gap heartbeat, endedAt = lastSeenAt at finalize). Co-Authored-By: Claude Opus 4.8 --- pkg/spxrpc/place_stream_live.go | 84 +++++++++++ pkg/statedb/queue_processor.go | 28 ++++ pkg/statedb/queue_processor_finalize_test.go | 148 +++++++++++++++++++ 3 files changed, 260 insertions(+) create mode 100644 pkg/statedb/queue_processor_finalize_test.go diff --git a/pkg/spxrpc/place_stream_live.go b/pkg/spxrpc/place_stream_live.go index b1c5fdb0..a3df1847 100644 --- a/pkg/spxrpc/place_stream_live.go +++ b/pkg/spxrpc/place_stream_live.go @@ -543,6 +543,19 @@ func (s *Server) handlePlaceStreamLiveStartLivestream(ctx context.Context, body livestream.CreatedAt = now livestream.LastSeenAt = &now + // End any prior un-ended livestream for this repo before creating the new + // one. Creating a new place.stream.livestream record (a new rkey) does not + // touch the prior record, so its idle-timeout finalize task — enqueued at + // that record's sync time and keyed to its URI — keeps ticking against a + // lastSeenAt that stops being refreshed once this new record becomes + // "latest". Ending the prior record here sets endedAt, which makes the + // finalize task's rec.EndedAt != nil early-skip fire instead of writing a + // stale endedAt later. Best-effort: a failure only logs, it must not block + // the new stream from starting. + if err := s.endPriorLivestream(ctx, session.DID, client); err != nil { + log.Error(ctx, "failed to end prior livestream before starting new one", "error", err) + } + if livestream.Thumb == nil { // Upload the user's current thumbnail to their PDS as the livestream image. var thumb *lexutil.LexBlob @@ -653,6 +666,77 @@ func (s *Server) handlePlaceStreamLiveStartLivestream(ctx context.Context, body }, nil } +// endPriorLivestream ends the streamer's latest livestream if it has not yet +// been ended. Called from startLivestream so that minting a new +// place.stream.livestream record supersedes the prior one: without this, the +// prior record's idle-timeout finalize task stays scheduled and keyed to its +// own URI, and once the new record becomes "latest" the prior record's +// lastSeenAt stops being refreshed — so the finalize task later writes a stale +// endedAt onto a record that was effectively replaced. Ending it here makes that +// task's rec.EndedAt != nil early-skip fire harmlessly. +// +// Mirrors the record-ending half of stopLivestream (getRecord for a fresh CID +// to swap on, set endedAt, putRecord) but is best-effort and never returns an +// error that blocks the new stream: callers log and continue. +func (s *Server) endPriorLivestream(ctx context.Context, repoDID string, client *oatproxy.XrpcClient) error { + prior, err := s.model.GetLatestLivestreamForRepo(repoDID) + if err != nil { + return fmt.Errorf("get latest livestream: %w", err) + } + if prior == nil || prior.Livestream == nil { + return nil + } + priorView, err := prior.ToLivestreamView() + if err != nil { + return fmt.Errorf("convert prior livestream to view: %w", err) + } + priorRec, ok := priorView.Record.Val.(*placestream.Livestream) + if !ok { + return fmt.Errorf("prior livestream is not a streamplace livestream") + } + if priorRec.EndedAt != nil { + // Already ended (e.g. by stopLivestream or an earlier finalize). The + // finalize task will skip it; nothing to do. + return nil + } + + aturi, err := syntax.ParseATURI(priorView.Uri) + if err != nil { + return fmt.Errorf("parse prior livestream URI: %w", err) + } + + // Fetch the current CID to swap on, so we don't clobber a concurrent + // update (and so the putRecord is rejected if the record changed). + var swapRecord *string + getOutput := comatproto.RepoGetRecord_Output{} + err = client.Do(ctx, xrpc.Query, "application/json", "com.atproto.repo.getRecord", map[string]any{ + "repo": repoDID, + "collection": "place.stream.livestream", + "rkey": aturi.RecordKey().String(), + }, nil, &getOutput) + if err != nil { + return fmt.Errorf("get prior livestream record: %w", err) + } + swapRecord = getOutput.Cid + + now := time.Now().UTC().Format(util.ISO8601) + priorRec.EndedAt = &now + + inp := comatproto.RepoPutRecord_Input{ + Collection: "place.stream.livestream", + Record: &lexutil.LexiconTypeDecoder{Val: priorRec}, + Rkey: aturi.RecordKey().String(), + Repo: repoDID, + SwapRecord: swapRecord, + } + var out comatproto.RepoPutRecord_Output + if err := client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &out); err != nil { + return fmt.Errorf("end prior livestream: %w", err) + } + log.Log(ctx, "ended prior livestream on startLivestream", "uri", priorView.Uri, "endedAt", now) + return nil +} + func (s *Server) handlePlaceStreamLiveStopLivestream(ctx context.Context, body *placestream.LiveStopLivestream_Input) (*placestream.LiveStopLivestream_Output, error) { now := time.Now().UTC().Format(util.ISO8601) session, _ := oatproxy.GetOAuthSession(ctx) diff --git a/pkg/statedb/queue_processor.go b/pkg/statedb/queue_processor.go index 2adf648e..927b2b2c 100644 --- a/pkg/statedb/queue_processor.go +++ b/pkg/statedb/queue_processor.go @@ -10,6 +10,7 @@ import ( "github.com/bluesky-social/indigo/api/bsky" "github.com/bluesky-social/indigo/atproto/syntax" lexutil "github.com/bluesky-social/indigo/lex/util" + "github.com/bluesky-social/indigo/util" "github.com/bluesky-social/indigo/xrpc" "golang.org/x/sync/errgroup" "gorm.io/gorm" @@ -332,6 +333,33 @@ func (state *StatefulDB) processFinalizeLivestreamTask(ctx context.Context, task log.Debug(ctx, "livestream is active, skipping finalization", "lastSeenAt", lastSeenTime) return nil } + // If this record is still the streamer's latest livestream, do NOT end it + // on a stale lastSeenAt alone. lastSeenAt only advances via the per-segment + // heartbeat (StreamSession.doUpdateLivestream), which is coupled to segment + // arrival and can lag behind actual ingestion — e.g. after an ingest gap + // long enough to tear down and restart the StreamSession, the new session's + // heartbeat may not land on this record before the idle timer fires. Ending + // here would set endedAt on the record the active stream is publishing + // under, taking the stream pre-live underneath a still-flowing ingest. + // + // Instead, reschedule the check for one more idle window: if the stream is + // truly abandoned the heartbeat stays frozen and we end it on the next + // pass; if it's a heartbeat-lag artifact, the heartbeat catches up and the + // rescheduled task hits the "active" early-return above. + latest, err := state.model.GetLatestLivestreamForRepo(livestream.RepoDID) + if err != nil { + return fmt.Errorf("failed to get latest livestream for repo: %w", err) + } + if latest != nil && latest.URI == livestream.URI { + rescheduledAt := time.Now().Add(time.Duration(*rec.IdleTimeoutSeconds) * time.Second).UTC() + rescheduledKey := fmt.Sprintf("finalize-livestream::%s::%s", livestream.URI, rescheduledAt.Format(util.ISO8601)) + _, err = state.EnqueueTask(ctx, TaskFinalizeLivestream, finalizeLivestreamTask, WithTaskKey(rescheduledKey), WithScheduledAt(rescheduledAt)) + if err != nil { + return fmt.Errorf("failed to reschedule finalize livestream task: %w", err) + } + log.Log(ctx, "livestream is latest for repo but lastSeenAt is stale; rescheduling finalize to let heartbeat catch up", "uri", livestream.URI, "lastSeenAt", lastSeenTime, "rescheduledAt", rescheduledAt) + return nil + } session, err := state.GetSessionByDID(livestream.RepoDID) if err != nil { return fmt.Errorf("failed to get session: %w", err) diff --git a/pkg/statedb/queue_processor_finalize_test.go b/pkg/statedb/queue_processor_finalize_test.go new file mode 100644 index 00000000..9edb1858 --- /dev/null +++ b/pkg/statedb/queue_processor_finalize_test.go @@ -0,0 +1,148 @@ +package statedb + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/streamplace" +) + +// marshalLivestream encodes a streamplace.Livestream record to the CBOR blob +// shape the model stores (the same bytes atproto sync decodes via +// lexutil.CborDecodeValue). +func marshalLivestream(t *testing.T, rec *streamplace.Livestream) []byte { + t.Helper() + var buf bytes.Buffer + require.NoError(t, rec.MarshalCBOR(&buf)) + return buf.Bytes() +} + +// seedLivestream inserts a place.stream.livestream row for did with the given +// lastSeenAt/endedAt/idleTimeoutSeconds, returning its URI. createdAgo sets the +// row's created_at (the column GetLatestLivestreamForRepo orders by) so callers +// can control which record is "latest". +func seedLivestream(t *testing.T, mod model.Model, did, rkey string, createdAgo time.Duration, rec *streamplace.Livestream) string { + t.Helper() + // ToLivestreamView dereferences ls.Repo.Handle, so the streamer needs a + // repo row. UpdateRepo upserts on PK (did), creating it if absent. + require.NoError(t, mod.UpdateRepo(&model.Repo{DID: did, Handle: "handle-" + rkey})) + uri := "at://" + did + "/place.stream.livestream/" + rkey + created := time.Now().Add(-createdAgo) + blob := marshalLivestream(t, rec) + require.NoError(t, mod.CreateLivestream(context.Background(), &model.Livestream{ + URI: uri, + CID: "bafy-" + rkey, + CreatedAt: created, + Livestream: &blob, + RepoDID: did, + })) + return uri +} + +func ptr[T any](v T) *T { return &v } + +// TestFinalizeLivestreamReschedulesWhenLatestButStale proves the guard added to +// processFinalizeLivestreamTask: when a livestream's lastSeenAt is older than +// its idleTimeoutSeconds but the record is still the streamer's latest, the +// task must NOT set endedAt (which would take the active stream pre-live). It +// must instead reschedule itself for one more idle window and return nil, so a +// heartbeat that's lagging behind actual ingestion gets a chance to catch up. +// +// This is the exact failure that ended Eli's active stream on 2026-07-06: a +// ~60s ingest gap froze lastSeenAt, the idle timer fired while the stream was +// still flowing, and endedAt was written onto the live record. +func TestFinalizeLivestreamReschedulesWhenLatestButStale(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + ctx := context.Background() + did := "did:plc:reschedule" + + // The record is "latest" (only one for this repo) and its lastSeenAt + // is well past the 300s idle timeout. + uri := seedLivestream(t, state.model, did, "latest", 1*time.Hour, &streamplace.Livestream{ + LexiconTypeID: "place.stream.livestream", + CreatedAt: time.Now().Add(-1 * time.Hour).Format(time.RFC3339), + LastSeenAt: ptr(time.Now().Add(-10 * time.Minute).Format(time.RFC3339)), + IdleTimeoutSeconds: ptr(int64(300)), + }) + + task := &AppTask{ID: 1, Type: TaskFinalizeLivestream, Payload: mustMarshal(t, FinalizeLivestreamTask{ + LivestreamURI: uri, + })} + + // Must not reach the PDS client (no session is configured), and must + // not error — it reschedules and returns nil. + err := state.processFinalizeLivestreamTask(ctx, task) + require.NoError(t, err, "stale-but-latest must reschedule, not error or end") + + // A rescheduled task must have been enqueued, keyed to the same URI. + tasks, err := state.ListTasks(ctx, TaskFilters{Type: TaskFinalizeLivestream, Limit: 10}) + require.NoError(t, err) + require.Len(t, tasks, 1, "exactly one rescheduled finalize task expected") + require.Contains(t, *tasks[0].TaskKey, "finalize-livestream::"+uri+"::") + require.NotNil(t, tasks[0].ScheduledAt, "rescheduled task must carry a future ScheduledAt") + require.True(t, tasks[0].ScheduledAt.After(time.Now()), "rescheduled task must run in the future") + + // And critically: the record must NOT have been ended. Re-read it from + // the repo and confirm endedAt is still unset. + ls, err := state.model.GetLivestream(uri) + require.NoError(t, err) + view, err := ls.ToLivestreamView() + require.NoError(t, err) + rec, ok := view.Record.Val.(*streamplace.Livestream) + require.True(t, ok) + require.Nil(t, rec.EndedAt, "endedAt must not be set on a stale-but-latest record") + }) +} + +// TestFinalizeLivestreamEndsSupersededRecord proves the complementary case: when +// a newer livestream exists for the repo, the stale older record is no longer +// "latest" and is safe to end — so the guard must NOT reschedule. Instead the +// task proceeds toward ending it (which here surfaces as a session-lookup error, +// since no PDS session is wired; what matters is that no reschedule was enqueued +// and the record wasn't protected by the latest-record guard). +func TestFinalizeLivestreamEndsSupersededRecord(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + ctx := context.Background() + did := "did:plc:superseded" + + // Older record: stale lastSeenAt, but a newer record will exist. + _ = seedLivestream(t, state.model, did, "old", 2*time.Hour, &streamplace.Livestream{ + LexiconTypeID: "place.stream.livestream", + CreatedAt: time.Now().Add(-2 * time.Hour).Format(time.RFC3339), + LastSeenAt: ptr(time.Now().Add(-10 * time.Minute).Format(time.RFC3339)), + IdleTimeoutSeconds: ptr(int64(300)), + }) + // Newer record: makes "old" no longer latest. + _ = seedLivestream(t, state.model, did, "new", 1*time.Minute, &streamplace.Livestream{ + LexiconTypeID: "place.stream.livestream", + CreatedAt: time.Now().Add(-1 * time.Minute).Format(time.RFC3339), + LastSeenAt: ptr(time.Now().Format(time.RFC3339)), + IdleTimeoutSeconds: ptr(int64(300)), + }) + + oldURI := "at://" + did + "/place.stream.livestream/old" + task := &AppTask{ID: 2, Type: TaskFinalizeLivestream, Payload: mustMarshal(t, FinalizeLivestreamTask{ + LivestreamURI: oldURI, + })} + + // No reschedule should be enqueued for a superseded record. The task + // proceeds past the guard and fails at session lookup (no PDS wired), + // which is the expected "didn't take the guard path" signal. + err := state.processFinalizeLivestreamTask(ctx, task) + require.Error(t, err, "superseded record should proceed to end (here: fail at session lookup), not reschedule") + + tasks, err := state.ListTasks(ctx, TaskFilters{Type: TaskFinalizeLivestream, Limit: 10}) + require.NoError(t, err) + require.Empty(t, tasks, "no reschedule task should be enqueued for a superseded record") + }) +} + +// smoke: the memory-mode StatefulDB used above needs config.DBURL only for the +// non-draft test bootstrap path; reference it so an unused import can't bite +// if these tests grow. +var _ = config.CLI{} -- 2.51.2