Something went wrong. Try again.
Something went wrong. Try again.
13 kB · 266 lines
Go
at commit d1dfd2ca
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267package statedb
import ( "bytes" "context" "testing" "time"
"github.com/streamplace/oatproxy/pkg/oatproxy" "github.com/stretchr/testify/require" "gorm.io/gorm" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/placestream")
// marshalLivestream encodes a placestream.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 *placestream.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 *placestream.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 }
// seedSession inserts a non-revoked OAuth session row for did, so the repo// counts as "has a local session on this node." The finalize guard reschedules// only when a local session exists (the heartbeat-lag scenario is real only// for repos this node is actively ingesting). Without this, the stale-but-latest// guard drops the task instead (firehose-observed, never connected here).func seedSession(t *testing.T, state *StatefulDB, did string) { t.Helper() require.NoError(t, state.CreateOAuthSession("jkt-"+did, &oatproxy.OAuthSession{ DID: did, DownstreamDPoPJKT: "jkt-" + did, }))}
// 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, AND// the repo has a local OAuth session (so heartbeat lag is possible), 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, &placestream.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)), })
// A local session exists, so the heartbeat-lag guard applies and // the task reschedules instead of ending or dropping. seedSession(t, state, did)
task := &AppTask{ID: 1, Type: TaskFinalizeLivestream, Payload: mustMarshal(t, FinalizeLivestreamTask{ LivestreamURI: uri, })}
// Must not reach the PDS client (no real PDS is configured, only // the session row), and must not error — it reschedules and // returns nil. err := state.processFinalizeLivestreamTask(ctx, task) require.NoError(t, err, "stale-but-latest with a session 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.(*placestream.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, &placestream.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, &placestream.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") })}
// TestFinalizeLivestreamDropsStaleLatestWithoutSession proves the fix for the// dev-environment log flood: when a stale-but-latest livestream has NO local// OAuth session (it arrived via firehose sync from an account that never// connected to this node), the task must be dropped — not rescheduled.//// Without a local session there is no heartbeat to wait for (doUpdateLivestream// only runs inside a local StreamSession) and no write access to end the record.// Rescheduling would respawn the task every idle window forever — the// rescheduled key embeds a fresh timestamp, so dedup never fires — flooding the// logs and growing the task table without bound.func TestFinalizeLivestreamDropsStaleLatestWithoutSession(t *testing.T) { WithAllDatabases(t, func(state *StatefulDB) { ctx := context.Background() did := "did:plc:firehose-only"
// 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, &placestream.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)), }) // Deliberately NO seedSession: this repo was only seen on the firehose.
// Create the task row so CompleteTask has something to mark done. createdTask, err := state.EnqueueTask(ctx, TaskFinalizeLivestream, FinalizeLivestreamTask{ LivestreamURI: uri, }, WithTaskKey("finalize-livestream::"+uri+"::initial")) require.NoError(t, err)
// Must not error, must not reschedule, must not reach the PDS client. // The task is dropped (completed) because there's no session and thus // no heartbeat-lag scenario to guard. err = state.processFinalizeLivestreamTask(ctx, createdTask) require.NoError(t, err, "stale-but-latest with no session must drop, not error or reschedule")
// No rescheduled (pending) task should have been enqueued. The // original task row is COMPLETED; filter to PENDING to assert no // reschedule was spawned. tasks, err := state.ListTasks(ctx, TaskFilters{Type: TaskFinalizeLivestream, Status: TaskStatusPending, Limit: 10}) require.NoError(t, err) require.Empty(t, tasks, "no reschedule task should be enqueued for a session-less stale-but-latest record")
// The record must NOT have been ended (no session = no write access). ls, err := state.model.GetLivestream(uri) require.NoError(t, err) view, err := ls.ToLivestreamView() require.NoError(t, err) rec, ok := view.Record.Val.(*placestream.Livestream) require.True(t, ok) require.Nil(t, rec.EndedAt, "endedAt must not be set without a local session to write it") })}
// TestFinalizeLivestreamSupersededWithoutSessionDropsNotErrors is a regression// guard: a superseded record with no session should still be droppable. In// practice superseded records proceed past the latest-record guard to the// session-lookup path, where GetSessionByDID returns gorm.ErrRecordNotFound.// This test documents that the superseded path still errors at session lookup// (it doesn't silently drop), confirming the no-session drop only applies to// the stale-but-latest branch.func TestFinalizeLivestreamSupersededWithoutSessionErrorsAtSessionLookup(t *testing.T) { WithAllDatabases(t, func(state *StatefulDB) { ctx := context.Background() did := "did:plc:superseded-nosession"
// Older record: stale, will be superseded. _ = seedLivestream(t, state.model, did, "old", 2*time.Hour, &placestream.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, &placestream.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)), }) // No session for this repo.
oldURI := "at://" + did + "/place.stream.livestream/old" task := &AppTask{ID: 2, Type: TaskFinalizeLivestream, Payload: mustMarshal(t, FinalizeLivestreamTask{ LivestreamURI: oldURI, })}
err := state.processFinalizeLivestreamTask(ctx, task) // Superseded records proceed past the guard to session lookup, which // fails with gorm.ErrRecordNotFound — the no-session drop only // applies to the stale-but-latest branch. require.Error(t, err, "superseded record without session should error at session lookup, not drop") require.ErrorIs(t, err, gorm.ErrRecordNotFound) })}
// 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{}