diff --git a/docs/HORIZONTAL_SCALING.md b/docs/HORIZONTAL_SCALING.md index 08e49bf..5d5131d 100644 --- a/docs/HORIZONTAL_SCALING.md +++ b/docs/HORIZONTAL_SCALING.md @@ -27,7 +27,7 @@ in that set, and the derived tables can mostly be ignored. | `hold_captain_records` | `io.atcr.hold.captain` | | `hold_crew_members` | `io.atcr.hold.crew` | | `scans` | `io.atcr.hold.scan` | -| `users` (identity columns) | DID resolution + `app.bsky.actor.profile` | +| `users` (all but `last_seen`) | DID resolution, `app.bsky.actor.profile`, `io.atcr.sailor.profile` | Stale reads here are self-correcting. A user who pushes an image and does not see it for a few seconds is a cosmetic problem; the next backfill fixes any @@ -46,19 +46,21 @@ divergence permanently. | `stripe_processed_events` | Idempotency ledger. Losing it means reprocessing Stripe events. | | `schema_migrations` | Migrations re-run against a database that already has them. | | `advisor_suggestions` | Regenerable, at AI cost. | +| `users.last_seen` | The only non-derived column in an otherwise derived table. | Self-healing, so effectively free to lose: `instance_leases`, `hold_crew_approvals`, `hold_crew_denials`, `jetstream_cursor` (costs a re-crawl), `labeler_cursor` + `taken_down_subjects` (replayable from the labeler from cursor 0). -### One trap in that split +### `users` is fully derived, including preferences -`users` is derived **except** for `oci_client` and `registry_domain`, which are -user preferences set from the settings form and never written to a PDS. Anything -that "rebuilds users from the PDS" wholesale would silently wipe them. Nothing -does today (the backfill only upserts identity columns), but a future cleanup -that truncates `users` would lose real user data. +`oci_client` and `registry_domain` look local but are not: both are fields on the +`io.atcr.sailor.profile` record. The settings form writes them to the user's PDS +and the local columns are a cache, refreshed by `ProcessSailorProfile`. Same for +`default_hold_did`. So the whole table can be rebuilt, preferences included. + +`last_seen` is the exception, and it is not derived from anything — see below. ## What is done @@ -137,11 +139,24 @@ instance (a lease) is likely simpler than making the ledger conflict-free. Not correctness, cost. Both are cheap to fix and worth doing before any remote primary carries production traffic: -- `UpdateUserLastSeen` runs per Jetstream **event** for cached users - (`processor.go:73`). One remote write per indexed record, for a timestamp that - only needs minute granularity. -- `DeviceStore.UpdateLastUsed` runs per `/auth/token` call, i.e. per docker - push/pull, in a detached goroutine with no bound (`device_store.go:363,408`). +- `UpdateUserLastSeen` ran per Jetstream **event** for cached users. Now + throttled to once per five minutes per user. +- `DeviceStore.UpdateLastUsed` ran per `/auth/token` call, i.e. per docker + push/pull. Now throttled the same way. + +### `last_seen` means "this user did something recently" + +It is the one column in `users` that nothing upstream can rebuild, and it was +being written by the wrong things. + +The backfill walks every historical record in the network. Stamping `last_seen` +there recorded *when the backfill ran*, for every user at once, on every run, +which destroys the only signal the column carries. It is now written on the two +paths that represent real activity: an interactive login, and a live commit event +observed on the firehose (which does mean the user just wrote a record). + +Anyone computing MAU from this column should know it was unreliable for every +run before this change. ### 5. Not a problem, contrary to earlier suspicion diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index f54f22e..c388e69 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -533,22 +533,62 @@ func InsertUserIfNotExists(db DBTX, user *User) error { return err } -// UpsertUser inserts or updates a user record +// UserUpsertOptions controls which columns an upsert overwrites on an existing +// row. The insert path always writes every column. +type UserUpsertOptions struct { + // UpdateAvatar overwrites a stored avatar. Leave false when the avatar + // fetch failed, so an empty string does not replace a good value. + UpdateAvatar bool + + // UpdateLastSeen stamps last_seen. Set it only when the user actually did + // something: an interactive login, or a live record write observed on the + // firehose. + // + // It is deliberately NOT set by the backfill. The backfill walks every + // historical record in the network, so stamping there records when the + // backfill ran, not when the user was last active, and it does so for every + // user at once. That makes last_seen useless as an activity signal, which is + // exactly what it is for. + UpdateLastSeen bool +} + +// UpsertUser inserts or updates a user record, stamping last_seen. Use it on +// paths that represent real user activity; see UpsertUserObserved for indexing. func UpsertUser(db DBTX, user *User) error { + return UpsertUserWithOptions(db, user, UserUpsertOptions{UpdateAvatar: true, UpdateLastSeen: true}) +} + +// UpsertUserObserved records identity seen while indexing, without claiming the +// user did anything. Handle, PDS endpoint and avatar are still corrected, which +// is why the backfill re-resolves rather than trusting a cache. +func UpsertUserObserved(db DBTX, user *User, updateAvatar bool) error { + return UpsertUserWithOptions(db, user, UserUpsertOptions{UpdateAvatar: updateAvatar, UpdateLastSeen: false}) +} + +// UpsertUserWithOptions inserts or updates a user record. +func UpsertUserWithOptions(db DBTX, user *User, opts UserUpsertOptions) error { // Clear handle from any other DID that currently holds it. // In ATProto, a handle belongs to exactly one DID at a time — // if a new DID claims this handle, the old association is stale. _, _ = db.Exec(`UPDATE users SET handle = did WHERE handle = ? AND did != ?`, user.Handle, user.DID) + updates := []string{ + "handle = excluded.handle", + "pds_endpoint = excluded.pds_endpoint", + } + if opts.UpdateAvatar { + updates = append(updates, "avatar = excluded.avatar") + } + if opts.UpdateLastSeen { + updates = append(updates, "last_seen = excluded.last_seen") + } + _, err := db.Exec(` INSERT INTO users (did, handle, pds_endpoint, avatar, last_seen) VALUES (?, ?, ?, ?, ?) ON CONFLICT(did) DO UPDATE SET - handle = excluded.handle, - pds_endpoint = excluded.pds_endpoint, - avatar = excluded.avatar, - last_seen = excluded.last_seen + `+strings.Join(updates, ",\n ")+` `, user.DID, user.Handle, user.PDSEndpoint, user.Avatar, user.LastSeen) return err } @@ -556,21 +596,7 @@ func UpsertUser(db DBTX, user *User) error { // UpsertUserIgnoreAvatar inserts or updates a user record, but preserves existing avatar on update // This is useful when avatar fetch fails, and we don't want to overwrite an existing avatar with empty string func UpsertUserIgnoreAvatar(db DBTX, user *User) error { - // Clear handle from any other DID that currently holds it. - // In ATProto, a handle belongs to exactly one DID at a time — - // if a new DID claims this handle, the old association is stale. - _, _ = db.Exec(`UPDATE users SET handle = did WHERE handle = ? AND did != ?`, - user.Handle, user.DID) - - _, err := db.Exec(` - INSERT INTO users (did, handle, pds_endpoint, avatar, last_seen) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(did) DO UPDATE SET - handle = excluded.handle, - pds_endpoint = excluded.pds_endpoint, - last_seen = excluded.last_seen - `, user.DID, user.Handle, user.PDSEndpoint, user.Avatar, user.LastSeen) - return err + return UpsertUserWithOptions(db, user, UserUpsertOptions{UpdateAvatar: false, UpdateLastSeen: true}) } // UpdateUserLastSeen updates only the last_seen timestamp for a user diff --git a/pkg/appview/jetstream/processor.go b/pkg/appview/jetstream/processor.go index e4a4200..50b0319 100644 --- a/pkg/appview/jetstream/processor.go +++ b/pkg/appview/jetstream/processor.go @@ -108,13 +108,19 @@ func (p *Processor) EnsureUser(ctx context.Context, did string) error { p.userCache.cache[did] = user } - // Upsert to database - // Use UpsertUser if we successfully fetched an avatar (to update existing users) - // Use UpsertUserIgnoreAvatar if fetch failed (to preserve existing avatars) - if avatarURL != "" { - return db.UpsertUser(p.db, user) - } - return db.UpsertUserIgnoreAvatar(p.db, user) + // Upsert to database. The avatar is only overwritten when we actually + // fetched one, so a failed fetch does not blank an existing avatar. + // + // last_seen is stamped only on the live path. p.useCache is true exactly for + // the Worker and false for the backfill, and the backfill walks every + // historical record in the network: stamping there would record when the + // backfill ran rather than when this user was last active, and would do it + // for every user at once. A live commit event, by contrast, means this user + // just wrote a record, which is real activity. + return db.UpsertUserWithOptions(p.db, user, db.UserUpsertOptions{ + UpdateAvatar: avatarURL != "", + UpdateLastSeen: p.useCache, + }) } // lastSeenInterval is how often a user's last_seen timestamp is actually diff --git a/pkg/appview/jetstream/processor_test.go b/pkg/appview/jetstream/processor_test.go index b1833e1..54c364e 100644 --- a/pkg/appview/jetstream/processor_test.go +++ b/pkg/appview/jetstream/processor_test.go @@ -1000,3 +1000,89 @@ func TestProcessAccount(t *testing.T) { t.Errorf("Deletion of non-existent user should not error, got: %v", err) } } + +// TestBackfillDoesNotStampLastSeen: last_seen means "this user did something +// recently", and the backfill is not the user doing something. +// +// It walks every historical record in the network, so stamping there records +// when the backfill ran rather than when the user was last active — and does it +// for every user at once, on every run. That makes the column useless as an +// activity signal, which is the only thing it is for. +func TestBackfillDoesNotStampLastSeen(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + const did = "did:plc:lastseen" + atproto.SetDirectory(&fakeDirectory{byDID: map[string]*identity.Identity{ + did: { + DID: syntax.DID(did), + Handle: syntax.Handle("lastseen.example.com"), + Services: map[string]identity.ServiceEndpoint{ + "atproto_pds": {Type: "AtprotoPersonalDataServer", URL: "https://pds.example.com"}, + }, + }, + }}) + + stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second) + if _, err := database.Exec(` + INSERT INTO users (did, handle, pds_endpoint, last_seen) + VALUES (?, 'lastseen.example.com', 'https://pds.example.com', ?) + `, did, stale); err != nil { + t.Fatalf("seed user: %v", err) + } + + // Backfill: useCache=false. + backfill := NewProcessor(database, false, nil) + if err := backfill.EnsureUser(context.Background(), did); err != nil { + t.Fatalf("backfill EnsureUser: %v", err) + } + + var after time.Time + if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&after); err != nil { + t.Fatalf("read last_seen: %v", err) + } + if !after.UTC().Truncate(time.Second).Equal(stale) { + t.Errorf("backfill moved last_seen from %v to %v; it now records when the backfill ran, "+ + "not when the user was active", stale, after) + } +} + +// TestLiveEventStampsLastSeen is the other half: a live commit means this user +// just wrote a record, which is activity worth recording. +func TestLiveEventStampsLastSeen(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + const did = "did:plc:lastseenlive" + atproto.SetDirectory(&fakeDirectory{byDID: map[string]*identity.Identity{ + did: { + DID: syntax.DID(did), + Handle: syntax.Handle("live.example.com"), + Services: map[string]identity.ServiceEndpoint{ + "atproto_pds": {Type: "AtprotoPersonalDataServer", URL: "https://pds.example.com"}, + }, + }, + }}) + + stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second) + if _, err := database.Exec(` + INSERT INTO users (did, handle, pds_endpoint, last_seen) + VALUES (?, 'live.example.com', 'https://pds.example.com', ?) + `, did, stale); err != nil { + t.Fatalf("seed user: %v", err) + } + + // Worker: useCache=true. + live := NewProcessor(database, true, nil) + if err := live.EnsureUser(context.Background(), did); err != nil { + t.Fatalf("live EnsureUser: %v", err) + } + + var after time.Time + if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&after); err != nil { + t.Fatalf("read last_seen: %v", err) + } + if !after.After(stale) { + t.Errorf("live event did not stamp last_seen: still %v", after) + } +}