appview: stop writing last_seen and last_used on every event

Neither is a correctness problem; both are round trips on hot paths for
timestamps nothing reads at that resolution.

UpdateUserLastSeen ran per Jetstream event for cached users, so once per indexed
record. DeviceStore.UpdateLastUsed ran per /auth/token call, so once per docker
push and pull including each layer's re-auth. Cheap against a local file, a
network round trip each against a remote primary, and the second sat on the
authentication path.

Both are now throttled to once per five minutes per subject. The MAU queries and
the admin views work in hours or days, so nothing loses meaning. The throttle
state is per-process and lost on restart, costing at most one extra write per
subject per boot; only the lease holder runs the consumer, so exactly one process
is doing the first of these at a time.

UpdateLastUsed stamps the throttle before writing rather than after, so a slow or
failing write cannot let every concurrent layer upload through to pile on more of
them.

Verified by disabling the throttle: 50 back-to-back calls then rewrite the
timestamp every time.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/HORIZONTAL_SCALING.md

@@ -0,0 +1,184 @@
+# Horizontally scaling the AppView
+
+Status of the work to make the AppView safe to run as N instances, and what is
+left before the database can move to local-write (Turso-style) embedded
+replicas.
+
+## The thing that decides everything else: most tables are derived
+
+The AppView database is mostly a **cache of ATProto records**. Jetstream and the
+backfill rebuild it from users' PDSes and from hold services. Losing a derived
+table costs a re-crawl, not data.
+
+A small set of tables is **authoritative**: nothing upstream can rebuild them,
+so staleness or loss is real loss. Almost every remaining scaling concern lives
+in that set, and the derived tables can mostly be ignored.
+
+### Derived — rebuilt by jetstream/backfill
+
+| Table | Source record |
+|---|---|
+| `manifests`, `layers`, `manifest_references` | `io.atcr.manifest` |
+| `tags` | `io.atcr.tag` |
+| `stars` | `io.atcr.sailor.star` |
+| `repo_pages` | `io.atcr.repo.page` |
+| `repository_annotations` | annotations on `io.atcr.manifest` |
+| `repository_stats`, `repository_stats_daily` | `io.atcr.hold.stats` |
+| `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` |
+
+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
+divergence permanently.
+
+### Authoritative — nothing upstream can rebuild these
+
+| Table | Cost if lost or read stale |
+|---|---|
+| `crypto_keys` | Catastrophic. Every registry JWT and OAuth client assertion becomes unverifiable. |
+| `oauth_sessions` | Every user must re-authenticate. Refresh tokens cannot be recovered. |
+| `ui_sessions` | Users logged out. |
+| `devices` | Every registered device must be re-enrolled; the secret is not recoverable. |
+| `pending_device_auth` | In-flight device logins fail. |
+| `webhooks` | User-created configuration, silently gone. |
+| `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. |
+
+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 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.
+
+## What is done
+
+Running N instances against one shared database is safe now.
+
+- **`instance_leases` + `pkg/appview/leases`.** Exactly one instance runs the
+  Jetstream consumer, backfill, labeler subscriber, cleanup sweep and billing
+  tier refresh. The consumer in particular *must* be a singleton: `StatsCache` is
+  per-process in-memory state whose aggregate is written to `repository_stats` as
+  an absolute value, so two consumers overwrite each other with partial sums, and
+  every webhook fires twice.
+- **OAuth session compare-and-swap.** Refresh tokens rotate on use, and the
+  per-DID mutex that serialized refreshes is in-process only. A second instance
+  refreshing the same account got `invalid_grant` and deleted the session out
+  from under the user. Writes now CAS on `oauth_sessions.rev`, and the delete
+  path checks whether the revision moved before destroying anything.
+- **Atomic crew denial counter.** Was a read-modify-write; concurrent denials
+  lost increments and the backoff escalated slower than configured.
+- **`crypto_keys` first-writer-wins.** Two instances booting against a fresh
+  database both generated a key and the loser kept its own in memory.
+- **Denial cache no longer wiped on every boot.** `DELETE FROM
+  hold_crew_denials` ran unconditionally at startup, so a rolling deploy wiped
+  the shared table once per instance.
+- **Node-independent keys.** `tags.id` dropped; `manifests.id` replaced by
+  `manifest_key`, derived from `(did, repository, digest)`. No rowid is allocated
+  by a node any more.
+- **Schema drift is checked**, both as a test (`schema.sql` vs the migrations)
+  and as a warning at boot.
+
+## What is left: read-after-write under local-write replicas
+
+None of the following is a problem today. With write-forwarding replicas every
+write goes to one primary, so all instances read a single consistent state.
+They become problems only if the database moves to **local-write** replicas,
+where each node writes locally and reconciles afterwards.
+
+Given the derived/authoritative split, the list is short.
+
+### 1. Session and device flows break visibly
+
+These are authoritative and read immediately after write, by a *different*
+instance than the one that wrote:
+
+- **`ui_sessions`** — log in on instance A, the next request is routed to B, B
+  does not have the session yet, user appears logged out.
+- **`pending_device_auth`** — A creates the pending row, the user approves on B,
+  the CLI polls C. If the poll interval is shorter than the sync interval the CLI
+  reports "still pending" after approval already happened, and may time out.
+- **`devices`** — enrol on A, first push authenticates against B.
+
+These need read-through-to-primary (or a forced sync) on the specific endpoints,
+not a general consistency guarantee. The set of endpoints is small: the OAuth
+callback, the device-code poll, and device authentication.
+
+### 2. The OAuth CAS stops being a CAS
+
+`oauth_sessions.rev` compare-and-swap assumes the `UPDATE ... WHERE rev = ?`
+either wins or loses against one authoritative row. Under local writes both
+nodes' updates succeed locally and conflict at reconciliation, where last-writer
+wins by default — exactly the clobber the CAS exists to prevent.
+
+This is the one place where local-write replication is genuinely incompatible
+with the current design rather than merely inconvenient. Options: keep OAuth
+sessions on a single-writer store, or move the per-DID lock to something with a
+real serialization point.
+
+### 3. `stripe_processed_events` needs a real barrier
+
+The whole point of the table is that an event is processed exactly once. Two
+nodes handling a redelivery concurrently would both find the row absent locally.
+Billing is behind a build tag and low volume, so pinning webhook handling to one
+instance (a lease) is likely simpler than making the ledger conflict-free.
+
+### 4. Write amplification on the hot path
+
+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`).
+
+### 5. Not a problem, contrary to earlier suspicion
+
+The second `?mode=ro` connection in `readonly.go` is fine in local-only mode:
+verified that a write through the read-write handle is immediately visible to the
+read-only one. Under embedded replicas it reads a file the replica connector is
+syncing beneath it, which has not been verified against a real remote, but the
+staleness that implies is already the documented expectation for that handle.
+
+## The labeler's `labels.id` is not the same problem
+
+`pkg/labeler` still uses `INTEGER PRIMARY KEY AUTOINCREMENT`, and should.
+
+That id is the **sequence number of the `com.atproto.label.subscribeLabels`
+stream**. Consumers use it as a resumption cursor (`GetLabelsSince` is
+`WHERE id > ? ORDER BY id ASC`), and `LatestSeq` is `MAX(id)`. Label negation
+ordering also depends on it (`l2.id > l1.id` decides which label supersedes
+which). A protocol stream sequence must be monotonic and totally ordered, which
+by definition requires a single allocator. A derived key would have no ordering
+at all, so the trick used for manifests does not transfer.
+
+That is not a scaling defect, because a labeler **is** a single logical
+publisher. The right shape is one writer with read replicas, not N writers. It
+is also a separate service with its own database and its own `data_dir`, so none
+of the AppView's storage decisions reach it.
+
+The one thing worth knowing: `pkg/labeler/config.go` exposes `LibsqlSyncURL`, so
+the labeler *can* be run as an embedded replica. If that ever became a
+local-write replica with two instances creating labels, both would allocate the
+same sequence number and consumers would silently miss labels — no error, just a
+gap where a takedown should have been. If the labeler ever needs HA, it needs a
+leader election like the AppView's, not a cleverer key.
+
+## Recommended order
+
+1. Throttle the two hot-path writes (§4). Useful now, independent of everything.
+2. Decide the sync model. Under write-forwarding, nothing else here is required.
+3. If moving to local-write: fix the session and device flows (§1), then resolve
+   OAuth sessions and the Stripe ledger (§2, §3), which may mean keeping those
+   tables on a single-writer store rather than making them conflict-free.

pkg/appview/db/device_store.go

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -46,8 +47,22 @@ type PendingAuthorization struct {
 // DeviceStore manages devices and pending authorizations with SQLite persistence
 type DeviceStore struct {
 	db *sql.DB
+
+	// lastUsedWrite records when each device's last_used column was last
+	// written, so the timestamp is refreshed at most once per
+	// deviceLastUsedInterval instead of on every authentication.
+	lastUsedWrite sync.Map
 }
 
+// deviceLastUsedInterval is how often a device's last_used timestamp is actually
+// written. It is shown in the UI as "last used" and read by the MAU queries,
+// neither of which needs better than minute resolution.
+//
+// Without it, every /auth/token call writes — so every docker push and pull,
+// including each layer's re-auth. Cheap against a local file; a network round
+// trip against a remote primary, on the authentication path.
+const deviceLastUsedInterval = 5 * time.Minute
+
 // NewDeviceStore creates a new SQLite-backed device store
 func NewDeviceStore(db *sql.DB) *DeviceStore {
 	return &DeviceStore{db: db}
@@ -477,13 +492,28 @@ func (s *DeviceStore) RevokeDevice(did, deviceID string) error {
 	return nil
 }
 
-// UpdateLastUsed updates the last used timestamp
+// UpdateLastUsed updates the last used timestamp, at most once per
+// deviceLastUsedInterval per device.
+//
+// Callers invoke this on every successful authentication. The throttle state is
+// per-process and lost on restart, which costs one extra write per device per
+// boot.
 func (s *DeviceStore) UpdateLastUsed(secretHash string) {
+	now := time.Now()
+	if prev, ok := s.lastUsedWrite.Load(secretHash); ok {
+		if last, ok := prev.(time.Time); ok && now.Sub(last) < deviceLastUsedInterval {
+			return
+		}
+	}
+	// Stamp before writing rather than after: a slow or failing write must not
+	// let every concurrent layer upload through to pile on more of them.
+	s.lastUsedWrite.Store(secretHash, now)
+
 	_, err := s.db.Exec(`
 		UPDATE devices
 		SET last_used = ?
 		WHERE secret_hash = ?
-	`, time.Now(), secretHash)
+	`, now, secretHash)
 
 	if err != nil {
 		slog.Warn("Failed to update device last used timestamp", "component", "device_store", "error", err)

pkg/appview/db/device_store_test.go

@@ -753,3 +753,90 @@ func TestDeviceSecretLookup_StableAndDistinct(t *testing.T) {
 		t.Errorf("expected 64 hex chars for sha256, got %d", len(a))
 	}
 }
+
+// TestUpdateLastUsedIsThrottled: callers invoke this on every successful
+// authentication, which means every docker push and pull, including each layer's
+// re-auth. Writing every time is a network round trip per call against a remote
+// primary, for a timestamp read at minute resolution at best.
+func TestUpdateLastUsedIsThrottled(t *testing.T) {
+	database := deviceThrottleDB(t)
+	store := NewDeviceStore(database)
+	hash := seedThrottleDevice(t, database, "dev-a", "hash-a")
+
+	store.UpdateLastUsed(hash)
+	var first sql.NullTime
+	if err := database.QueryRow(`SELECT last_used FROM devices WHERE id = ?`, "dev-a").Scan(&first); err != nil {
+		t.Fatalf("read last_used: %v", err)
+	}
+	if !first.Valid {
+		t.Fatal("first call did not write last_used")
+	}
+
+	// Force a value the next write would visibly change, then hammer it.
+	marker := first.Time.Add(-time.Hour).UTC().Truncate(time.Second)
+	if _, err := database.Exec(`UPDATE devices SET last_used = ? WHERE id = ?`, marker, "dev-a"); err != nil {
+		t.Fatalf("set marker: %v", err)
+	}
+	for range 50 {
+		store.UpdateLastUsed(hash)
+	}
+
+	var after sql.NullTime
+	if err := database.QueryRow(`SELECT last_used FROM devices WHERE id = ?`, "dev-a").Scan(&after); err != nil {
+		t.Fatalf("read last_used: %v", err)
+	}
+	if !after.Time.UTC().Truncate(time.Second).Equal(marker) {
+		t.Error("last_used was rewritten during 50 back-to-back calls; the throttle is not holding")
+	}
+}
+
+// TestUpdateLastUsedThrottlesPerDevice: one busy device must not suppress
+// another device's first write.
+func TestUpdateLastUsedThrottlesPerDevice(t *testing.T) {
+	database := deviceThrottleDB(t)
+	store := NewDeviceStore(database)
+	hashA := seedThrottleDevice(t, database, "dev-a", "hash-a")
+	hashB := seedThrottleDevice(t, database, "dev-b", "hash-b")
+
+	store.UpdateLastUsed(hashA)
+	store.UpdateLastUsed(hashB)
+
+	for _, id := range []string{"dev-a", "dev-b"} {
+		var ts sql.NullTime
+		if err := database.QueryRow(`SELECT last_used FROM devices WHERE id = ?`, id).Scan(&ts); err != nil {
+			t.Fatalf("read last_used for %s: %v", id, err)
+		}
+		if !ts.Valid {
+			t.Errorf("device %s never got a last_used write", id)
+		}
+	}
+}
+
+func seedThrottleDevice(t *testing.T, database *sql.DB, id, secretHash string) string {
+	t.Helper()
+	if _, err := database.Exec(`
+		INSERT INTO devices (id, did, handle, name, secret_hash, created_at)
+		VALUES (?, 'did:plc:throttle', 'throttle.example.com', ?, ?, ?)
+	`, id, id, secretHash, time.Now()); err != nil {
+		t.Fatalf("seed device %s: %v", id, err)
+	}
+	return secretHash
+}
+
+func deviceThrottleDB(t *testing.T) *sql.DB {
+	t.Helper()
+	database, err := InitDB(":memory:", LibsqlConfig{})
+	if err != nil {
+		t.Fatalf("InitDB: %v", err)
+	}
+	t.Cleanup(func() { database.Close() })
+	if err := UpsertUser(database, &User{
+		DID:         "did:plc:throttle",
+		Handle:      "throttle.example.com",
+		PDSEndpoint: "https://pds.example.com",
+		LastSeen:    time.Now(),
+	}); err != nil {
+		t.Fatalf("UpsertUser: %v", err)
+	}
+	return database
+}

pkg/appview/jetstream/processor.go

@@ -56,7 +56,8 @@ func NewProcessor(database db.DBTX, useCache bool, statsCache *StatsCache) *Proc
 
 	if useCache {
 		p.userCache = &UserCache{
-			cache: make(map[string]*db.User),
+			cache:         make(map[string]*db.User),
+			lastSeenWrite: make(map[string]time.Time),
 		}
 	}
 
@@ -70,7 +71,7 @@ func (p *Processor) EnsureUser(ctx context.Context, did string) error {
 	// a user's identity won't change, so the cache hit is safe.
 	if p.useCache && p.userCache != nil {
 		if _, ok := p.userCache.cache[did]; ok {
-			return db.UpdateUserLastSeen(p.db, did)
+			return p.touchLastSeen(did)
 		}
 	}
 	// No cache early-return: always re-resolve identity so stale handles
@@ -116,6 +117,36 @@ func (p *Processor) EnsureUser(ctx context.Context, did string) error {
 	return db.UpsertUserIgnoreAvatar(p.db, user)
 }
 
+// lastSeenInterval is how often a user's last_seen timestamp is actually
+// written. Everything that reads it (the MAU queries, the admin views) works in
+// hours or days, so writing it more often than this buys nothing.
+const lastSeenInterval = 5 * time.Minute
+
+// touchLastSeen updates a user's last_seen, at most once per lastSeenInterval.
+//
+// This used to write on every Jetstream event for a cached user, which is one
+// database round trip per indexed record for a timestamp nobody reads at that
+// resolution. Cheap against a local file; not cheap against a remote primary,
+// where it is a network round trip in the middle of the indexing hot path.
+//
+// The throttle state is per-process and lost on restart, which costs at most one
+// extra write per user per boot. Only the lease holder runs the consumer, so
+// there is exactly one process doing this at a time.
+func (p *Processor) touchLastSeen(did string) error {
+	if p.userCache != nil {
+		if last, ok := p.userCache.lastSeenWrite[did]; ok && time.Since(last) < lastSeenInterval {
+			return nil
+		}
+	}
+	if err := db.UpdateUserLastSeen(p.db, did); err != nil {
+		return err
+	}
+	if p.userCache != nil {
+		p.userCache.lastSeenWrite[did] = time.Now()
+	}
+	return nil
+}
+
 // EnsureUserExists ensures a user row exists in the database without updating it.
 // Used by non-profile collections to avoid unnecessary writes during backfill.
 // If the user doesn't exist, resolves identity and inserts with ON CONFLICT DO NOTHING.

pkg/appview/jetstream/worker.go

@@ -19,9 +19,14 @@ import (
 	"github.com/klauspost/compress/zstd"
 )
 
-// UserCache caches DID -> handle/PDS mappings to avoid repeated lookups
+// UserCache caches DID -> handle/PDS mappings to avoid repeated lookups.
+//
+// lastSeenWrite records when each DID's last_seen column was last written, so
+// the timestamp is refreshed at most once per lastSeenInterval rather than on
+// every event. See Processor.EnsureUser.
 type UserCache struct {
-	cache map[string]*db.User
+	cache         map[string]*db.User
+	lastSeenWrite map[string]time.Time
 }
 
 // EventCallback is called for each processed event