mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-09-26 04:04:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

appview: stop the backfill claiming every user was just active

Browse Source 
last_seen means "this user did something recently". The backfill walks every
historical record in the network, so stamping it there recorded when the
backfill ran, not when the user was active — for every user at once, on every
run. That destroys the only signal the column carries, and it is the one column
in users that nothing upstream can rebuild.

It is now written on the two paths that represent real activity: an interactive
login, and a live commit event on the firehose, which does mean the user just
wrote a record. The backfill still corrects handle, PDS endpoint and avatar,
which is why it re-resolves rather than trusting a cache; it just no longer
claims the user was present.

UpsertUser grows an options form rather than a fourth named variant, since the
avatar and last_seen decisions are independent and all four combinations occur.

Anyone computing MAU from this column should know it was unreliable for every
backfill run before this change.

Also corrects docs/HORIZONTAL_SCALING.md, which claimed oci_client and
registry_domain were local-only preferences. They are fields on
io.atcr.sailor.profile: settings writes them to the user's PDS and
ProcessSailorProfile refreshes the local cache. users is fully derived apart
from last_seen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Evan Jarrett
2026-08-12 09:24:26 -05:00
co-authored by Claude Opus 5
parent 13edb7184d
commit 4c04983e23
4 changed files with 172 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/HORIZONTAL_SCALING.md
+27 -12
View File
@@ -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
pkg/appview/db/queries.go
+46 -20
View File
@@ -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
pkg/appview/jetstream/processor.go
+13 -7
View File
@@ -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
pkg/appview/jetstream/processor_test.go
+86
View File
@@ -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)
}
}