diff --git a/pkg/appview/db/device_store_test.go b/pkg/appview/db/device_store_test.go index 8b7c771..598d270 100644 --- a/pkg/appview/db/device_store_test.go +++ b/pkg/appview/db/device_store_test.go @@ -1,10 +1,12 @@ package db import ( + "bytes" "context" "database/sql" "errors" "fmt" + "log/slog" "strings" "testing" "time" @@ -840,3 +842,45 @@ func deviceThrottleDB(t *testing.T) *sql.DB { } return database } + +// TestUpdateLastUsedThrottlesEvenWhenTheWriteFails pins the statement order +// 13edb71 calls out: the throttle is stamped before the write, not after. +// +// The reason is the authentication path. UpdateLastUsed runs on every +// successful auth, which during a push means once per layer, concurrently. If +// the timestamp were recorded only after a successful write, then a write that +// is slow or failing would let every one of those callers through to queue +// another one behind it — the pile-up is worst exactly when the database is +// least able to absorb it. +// +// A failing write makes that ordering observable without timing anything: with +// the stamp after the write, every call retries; with it before, only the first +// does. The failure is counted through the warning the function logs, since a +// dropped table leaves no row to inspect. +func TestUpdateLastUsedThrottlesEvenWhenTheWriteFails(t *testing.T) { + database := deviceThrottleDB(t) + store := NewDeviceStore(database) + hash := seedThrottleDevice(t, database, "dev-fail", "hash-fail") + + // Take the table away so every UPDATE errors. + if _, err := database.Exec(`DROP TABLE devices`); err != nil { + t.Fatalf("drop devices: %v", err) + } + + var logged bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logged, &slog.HandlerOptions{Level: slog.LevelWarn}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + const calls = 10 + for range calls { + store.UpdateLastUsed(hash) + } + + got := strings.Count(logged.String(), "Failed to update device last used timestamp") + if got != 1 { + t.Errorf("failing write was attempted %d times across %d calls, want 1; "+ + "the throttle is stamped after the write, so a failing or slow write "+ + "lets every concurrent caller pile on another attempt", got, calls) + } +} diff --git a/pkg/appview/jetstream/processor_test.go b/pkg/appview/jetstream/processor_test.go index 54c364e..7d9d0c9 100644 --- a/pkg/appview/jetstream/processor_test.go +++ b/pkg/appview/jetstream/processor_test.go @@ -1086,3 +1086,96 @@ func TestLiveEventStampsLastSeen(t *testing.T) { t.Errorf("live event did not stamp last_seen: still %v", after) } } + +// TestTouchLastSeenIsThrottled covers the other half of 13edb71. +// +// That commit throttles two writes: DeviceStore.UpdateLastUsed and this one. +// Only the device side got a test. touchLastSeen is the hotter of the two — it +// ran once per indexed record, so a busy firehose meant a database round trip +// per event for a timestamp read in hours or days. Deleting the throttle here +// leaves every existing test green. +func TestTouchLastSeenIsThrottled(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + const did = "did:plc:touchthrottle" + 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 (?, 'touch.example.com', 'https://pds.example.com', ?) + `, did, stale); err != nil { + t.Fatalf("seed user: %v", err) + } + + // useCache=true is the live worker, and the only configuration that has a + // cache to throttle against. + live := NewProcessor(database, true, nil) + + if err := live.touchLastSeen(did); err != nil { + t.Fatalf("first touchLastSeen: %v", err) + } + var first time.Time + if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, did).Scan(&first); err != nil { + t.Fatalf("read last_seen: %v", err) + } + if first.UTC().Truncate(time.Second).Equal(stale) { + t.Fatal("first call did not write last_seen") + } + + // Put a value there that any further write would visibly change, then + // hammer it the way a firehose burst would. + marker := first.Add(-time.Hour).UTC().Truncate(time.Second) + if _, err := database.Exec(`UPDATE users SET last_seen = ? WHERE did = ?`, marker, did); err != nil { + t.Fatalf("set marker: %v", err) + } + for range 50 { + if err := live.touchLastSeen(did); err != nil { + t.Fatalf("throttled touchLastSeen: %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(marker) { + t.Error("last_seen was rewritten during 50 back-to-back events; the throttle is not holding") + } +} + +// TestTouchLastSeenThrottlesPerUser: one busy account must not suppress +// another's first write, which is what a shared timestamp would do. +func TestTouchLastSeenThrottlesPerUser(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + const didA = "did:plc:busyuser" + const didB = "did:plc:quietuser" + stale := time.Now().Add(-90 * 24 * time.Hour).UTC().Truncate(time.Second) + for _, did := range []string{didA, didB} { + if _, err := database.Exec(` + INSERT INTO users (did, handle, pds_endpoint, last_seen) + VALUES (?, ?, 'https://pds.example.com', ?) + `, did, did+".example.com", stale); err != nil { + t.Fatalf("seed %s: %v", did, err) + } + } + + live := NewProcessor(database, true, nil) + for range 10 { + if err := live.touchLastSeen(didA); err != nil { + t.Fatalf("touchLastSeen A: %v", err) + } + } + if err := live.touchLastSeen(didB); err != nil { + t.Fatalf("touchLastSeen B: %v", err) + } + + var b time.Time + if err := database.QueryRow(`SELECT last_seen FROM users WHERE did = ?`, didB).Scan(&b); err != nil { + t.Fatalf("read last_seen: %v", err) + } + if b.UTC().Truncate(time.Second).Equal(stale) { + t.Error("a busy account's writes suppressed a quiet account's first write") + } +}