diff --git a/config-appview.example.yaml b/config-appview.example.yaml index eb319bb..7176efe 100644 --- a/config-appview.example.yaml +++ b/config-appview.example.yaml @@ -56,6 +56,14 @@ health: cache_ttl: 15m0s # How often to refresh hold health checks. check_interval: 15m0s +# Leader election for background workers. Required when running more than one AppView instance. +leases: + # Elect a single instance to run background workers. Required when running more than one AppView instance. + enabled: true + # How long a lease survives without renewal before another instance may take it. Must exceed any plausible clock skew between instances. + ttl: 1m0s + # How often the holder renews its lease, and how often waiting instances retry. Must be well under ttl. + renew_interval: 20s # ATProto Jetstream event stream settings. jetstream: # Jetstream WebSocket endpoints, tried in order on failure. diff --git a/deploy/upcloud/configs/appview.yaml.tmpl b/deploy/upcloud/configs/appview.yaml.tmpl index e48b4ce..e62e49d 100644 --- a/deploy/upcloud/configs/appview.yaml.tmpl +++ b/deploy/upcloud/configs/appview.yaml.tmpl @@ -28,6 +28,12 @@ ui: health: cache_ttl: 15m0s check_interval: 15m0s +# Elects one instance to run the Jetstream consumer, backfill, labeler and +# cleanup workers. Harmless on a single instance; required before scaling out. +leases: + enabled: true + ttl: 1m0s + renew_interval: 20s jetstream: urls: - wss://jetstream2.us-west.bsky.network/subscribe diff --git a/pkg/appview/config.go b/pkg/appview/config.go index 41fe578..39a52d4 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -31,6 +31,7 @@ type Config struct { Server ServerConfig `yaml:"server" comment:"HTTP server and identity settings."` UI UIConfig `yaml:"ui" comment:"Web UI settings."` Health HealthConfig `yaml:"health" comment:"Health check and cache settings."` + Leases LeasesConfig `yaml:"leases" comment:"Leader election for background workers. Required when running more than one AppView instance."` Jetstream JetstreamConfig `yaml:"jetstream" comment:"ATProto Jetstream event stream settings."` Auth AuthConfig `yaml:"auth" comment:"JWT authentication settings."` CredentialHelper CredentialHelperConfig `yaml:"credential_helper" comment:"Credential helper download settings."` @@ -106,6 +107,27 @@ type HealthConfig struct { CheckInterval time.Duration `yaml:"check_interval" comment:"How often to refresh hold health checks."` } +// LeasesConfig controls leader election for background workers. +// +// The Jetstream consumer, backfill worker, labeler subscribers and cleanup loop +// must run on exactly one instance. Running two Jetstream consumers is not +// merely wasteful: each aggregates hold stats in process memory and writes the +// result to repository_stats as an absolute value, so two consumers overwrite +// one another with partial sums, and every webhook fires twice. +type LeasesConfig struct { + // Enabled turns on database leader election. Leave it on unless you have a + // specific reason not to; the cost is one small table and a renewal query + // every renew_interval. + Enabled bool `yaml:"enabled" comment:"Elect a single instance to run background workers. Required when running more than one AppView instance."` + + // TTL is how long a lease survives without renewal. + TTL time.Duration `yaml:"ttl" comment:"How long a lease survives without renewal before another instance may take it. Must exceed any plausible clock skew between instances."` + + // RenewInterval is how often the holder extends its lease, and how often a + // waiting instance retries. + RenewInterval time.Duration `yaml:"renew_interval" comment:"How often the holder renews its lease, and how often waiting instances retry. Must be well under ttl."` +} + // JetstreamConfig defines ATProto Jetstream settings type JetstreamConfig struct { // Jetstream WebSocket endpoints, tried in order on failure. @@ -219,6 +241,14 @@ func setDefaults(v *viper.Viper) { v.SetDefault("health.cache_ttl", "15m") v.SetDefault("health.check_interval", "15m") + // Lease defaults. On by default: a single instance is unaffected (it simply + // always wins its own leases), while an operator who scales out without + // reading the docs still gets correct behavior rather than duplicate + // Jetstream consumers silently corrupting repository_stats. + v.SetDefault("leases.enabled", true) + v.SetDefault("leases.ttl", "60s") + v.SetDefault("leases.renew_interval", "20s") + // Jetstream defaults v.SetDefault("jetstream.urls", []string{ "wss://jetstream2.us-west.bsky.network/subscribe", diff --git a/pkg/appview/db/readonly.go b/pkg/appview/db/readonly.go index 4343957..b3fc4ce 100644 --- a/pkg/appview/db/readonly.go +++ b/pkg/appview/db/readonly.go @@ -56,26 +56,38 @@ func InitializeDatabase(dbPath string, cfg LibsqlConfig) (*sql.DB, *sql.DB, *Ses // Create session store sessionStore := NewSessionStore(database) - // Start cleanup goroutines - go func() { - ticker := time.NewTicker(1 * time.Hour) - defer ticker.Stop() - for range ticker.C { - ctx := context.Background() - - // Cleanup UI sessions - sessionStore.Cleanup() - - // Cleanup OAuth sessions (older than 30 days) - oauthStore := NewOAuthStore(database) - oauthStore.CleanupOldSessions(ctx, 30*24*time.Hour) - oauthStore.CleanupExpiredAuthRequests(ctx) - - // Cleanup device pending auths - deviceStore := NewDeviceStore(database) - deviceStore.CleanupExpired() - } - }() - + // The periodic cleanup loop used to start here. It now runs under a lease + // so it executes on one instance rather than all of them; see + // RunPeriodicCleanup and AppViewServer.startCleanupWorker. return database, readOnlyDB, sessionStore } + +// CleanupExpiredRecords deletes expired UI sessions, old OAuth sessions, +// expired OAuth authorization requests and stale pending device authorizations. +// Every statement is an idempotent DELETE, so running it twice is harmless. +func CleanupExpiredRecords(ctx context.Context, database *sql.DB, sessionStore *SessionStore) { + sessionStore.Cleanup() + + oauthStore := NewOAuthStore(database) + oauthStore.CleanupOldSessions(ctx, 30*24*time.Hour) + oauthStore.CleanupExpiredAuthRequests(ctx) + + NewDeviceStore(database).CleanupExpired() +} + +// RunPeriodicCleanup runs CleanupExpiredRecords every interval until ctx is +// cancelled. It always returns ctx.Err(), so a lease manager treats a lost lease +// or a shutdown as "stopped" rather than "finished". +func RunPeriodicCleanup(ctx context.Context, database *sql.DB, sessionStore *SessionStore, interval time.Duration) error { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + CleanupExpiredRecords(ctx, database, sessionStore) + } + } +} diff --git a/pkg/appview/leases/manager.go b/pkg/appview/leases/manager.go index 48e7f10..918c398 100644 --- a/pkg/appview/leases/manager.go +++ b/pkg/appview/leases/manager.go @@ -27,6 +27,7 @@ import ( "fmt" "log/slog" "os" + "sync" "time" "atcr.io/pkg/appview/db" @@ -62,6 +63,10 @@ type Manager struct { db *sql.DB holderID string cfg Config + + // wg tracks workers started via Go, so Wait can block on their release + // during shutdown. + wg sync.WaitGroup } // NewManager returns a Manager identified by a holder ID unique to this process. @@ -113,6 +118,44 @@ func newHolderID() string { // instance that is not the leader. var errNotAcquired = errors.New("leases: not acquired") +// Go starts a leased worker in the background and registers it with the +// manager, so Wait can block on it during shutdown. +// +// Prefer this over `go m.Run(...)`. The registration has to happen on the +// caller's goroutine: doing it inside the spawned one races with Wait, which +// could then return before the worker has even started, let alone released its +// lease. +func (m *Manager) Go(ctx context.Context, name string, fn func(context.Context) error) { + m.wg.Add(1) + go func() { + defer m.wg.Done() + m.Run(ctx, name, fn) + }() +} + +// Wait blocks until every worker started with Go has stopped and released its +// lease, or until timeout elapses. +// +// Call this during shutdown, after cancelling the worker context. Without it the +// process exits while the release is still in flight, so the lease survives +// until its TTL lapses and the replacement instance sits idle for up to a full +// TTL. That is the difference between a rolling deploy pausing indexing for a +// second and pausing it for a minute. +func (m *Manager) Wait(timeout time.Duration) { + done := make(chan struct{}) + go func() { + m.wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(timeout): + slog.Warn("Timed out waiting for leased workers to stop; their leases will expire on their own", + "component", "leases", "timeout", timeout, "ttl", m.cfg.TTL) + } +} + // Run executes fn under the named lease, blocking until ctx is cancelled or fn // reports it has finished. // diff --git a/pkg/appview/leases/manager_test.go b/pkg/appview/leases/manager_test.go index e1746c1..b66a7ba 100644 --- a/pkg/appview/leases/manager_test.go +++ b/pkg/appview/leases/manager_test.go @@ -200,6 +200,88 @@ func TestReleaseOnExitAllowsImmediateTakeover(t *testing.T) { } } +// TestWaitBlocksUntilLeaseReleased pins the shutdown ordering. +// +// Regression test for a real bug: the workers were started with `go m.Run(...)` +// and the process exited immediately after cancelling their context, so it was +// gone before the release landed. The lease then survived until its TTL lapsed +// and the replacement instance sat idle for up to a minute. Two instances under +// a rolling deploy showed it; no unit test did. +func TestWaitBlocksUntilLeaseReleased(t *testing.T) { + database := testDB(t) + + m := NewManager(database, fastConfig()) + ctx, cancel := context.WithCancel(context.Background()) + + started := make(chan struct{}) + m.Go(ctx, "worker", func(workerCtx context.Context) error { + close(started) + <-workerCtx.Done() + return workerCtx.Err() + }) + + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("worker never started") + } + + cancel() + m.Wait(5 * time.Second) + + // Wait has returned, so the release must already have happened. Any expiry + // still in the future would mean the replacement has to wait it out. + lease, err := db.GetLease(database, "worker") + if err != nil { + t.Fatalf("GetLease: %v", err) + } + if lease == nil { + t.Fatal("lease row disappeared") + } + if lease.ExpiresAt.After(time.Now()) { + t.Errorf("Wait returned before the lease was released; it still expires at %v", lease.ExpiresAt) + } +} + +// TestWaitGivesUpOnAHungWorker: shutdown must not hang forever because one +// worker ignores its context. The lease expires on its own in that case. +func TestWaitGivesUpOnAHungWorker(t *testing.T) { + database := testDB(t) + + m := NewManager(database, fastConfig()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + release := make(chan struct{}) + t.Cleanup(func() { close(release) }) + + started := make(chan struct{}) + m.Go(ctx, "hung", func(context.Context) error { + close(started) + <-release // deliberately ignores the worker context + return nil + }) + + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("worker never started") + } + + cancel() + done := make(chan struct{}) + go func() { + m.Wait(200 * time.Millisecond) + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Wait did not give up on a worker that ignores its context") + } +} + // TestCompletedWorkReleasesAndStops: a one-shot worker returning nil means the // job is done. Run must not loop and re-run it forever. func TestCompletedWorkReleasesAndStops(t *testing.T) { diff --git a/pkg/appview/server.go b/pkg/appview/server.go index d173271..c73e753 100644 --- a/pkg/appview/server.go +++ b/pkg/appview/server.go @@ -27,6 +27,7 @@ import ( "atcr.io/pkg/appview/holdhealth" "atcr.io/pkg/appview/jetstream" appviewlabeler "atcr.io/pkg/appview/labeler" + "atcr.io/pkg/appview/leases" "atcr.io/pkg/appview/middleware" "atcr.io/pkg/appview/readme" "atcr.io/pkg/appview/registryauth" @@ -111,6 +112,9 @@ type AppViewServer struct { // WebhookDispatcher dispatches scan webhooks (stored in appview DB). WebhookDispatcher *webhooks.Dispatcher + // Leases elects a single instance to run each background worker. + Leases *leases.Manager + // Private fields for lifecycle management oauthHooks []OAuthPostAuthHook tokenHooks []TokenPostAuthHook @@ -186,6 +190,23 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, s.healthWorker.Start(workerCtx) slog.Info("Hold health worker started", "startup_delay", startupDelay, "refresh_interval", cfg.Health.CheckInterval, "cache_ttl", cfg.Health.CacheTTL) + // Leader election for the singleton background workers. The health worker + // above is deliberately not among them: it only refreshes a cache that each + // instance needs locally, so running it everywhere is correct. + s.Leases = leases.NewManager(s.Database, leases.Config{ + Enabled: cfg.Leases.Enabled, + TTL: cfg.Leases.TTL, + RenewInterval: cfg.Leases.RenewInterval, + }) + slog.Info("Lease manager initialized", + "component", "leases", "enabled", cfg.Leases.Enabled, "holder", s.Leases.HolderID()) + + // Session, OAuth and device-flow cleanup. Idempotent DELETEs, so running it + // on every instance would be harmless but wasteful; more importantly it + // belongs with the other leased workers rather than buried in the database + // constructor, where it had no way to reach the lease manager. + s.startCleanupWorker(workerCtx) + // Initialize OAuth components slog.Info("Initializing OAuth components") @@ -296,7 +317,13 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, return nil, fmt.Errorf("billing is enabled but STRIPE_WEBHOOK_SECRET is not set; refusing to start with a forgeable webhook endpoint") } slog.Info("Billing enabled", "appview_did", appviewDID, "managed_holds", len(cfg.Server.ManagedHolds)) - go s.BillingManager.RefreshHoldTiers() + // Leased: RefreshHoldTiers writes tier state derived from Stripe, and + // several instances refreshing the same holds concurrently would race + // on those writes for no benefit. + s.Leases.Go(workerCtx, db.LeaseBillingTiers, func(context.Context) error { + s.BillingManager.RefreshHoldTiers() + return nil // one-shot; the lease is released when it returns + }) } // Create webhook dispatcher @@ -314,15 +341,26 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, middleware.SetGlobalWebhookDispatcher(s.WebhookDispatcher) // Initialize Jetstream workers - s.initializeJetstream() + s.initializeJetstream(workerCtx) - // Initialize labeler subscriber + // Initialize labeler subscriber. Leased per labeler DID, matching the + // per-src cursor in labeler_cursor: two subscribers on one labeler would + // each advance the same cursor and double-apply every takedown. if cfg.Labeler.DID != "" { - sub := appviewlabeler.SubscriberFromConfig(cfg.Labeler.DID, s.Database) - if sub != nil { + labelerDID := cfg.Labeler.DID + s.Leases.Go(workerCtx, db.LeaseLabeler(labelerDID), func(ctx context.Context) error { + // Built inside the worker because Stop closes a channel and cannot + // be called twice; a retry after a lost lease needs a fresh one. + sub := appviewlabeler.SubscriberFromConfig(labelerDID, s.Database) + if sub == nil { + return fmt.Errorf("labeler subscriber unavailable for %q", labelerDID) + } sub.Start() - slog.Info("Labeler subscriber started", "labeler", cfg.Labeler.DID) - } + slog.Info("Labeler subscriber started", "labeler", labelerDID) + <-ctx.Done() + sub.Stop() + return ctx.Err() + }) } // Create main chi router @@ -732,6 +770,7 @@ func (s *AppViewServer) ServeWithListener(listener net.Listener) error { if s.workerCancel != nil { s.workerCancel() } + s.waitForLeasedWorkers() shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -745,6 +784,7 @@ func (s *AppViewServer) ServeWithListener(listener net.Listener) error { if s.workerCancel != nil { s.workerCancel() } + s.waitForLeasedWorkers() logging.Shutdown() if err != nil { return fmt.Errorf("server error: %w", err) @@ -926,8 +966,39 @@ func (s *AppViewServer) handleDIDDocument(w http.ResponseWriter, r *http.Request } } +// waitForLeasedWorkers gives the leased background workers a moment to stop and +// release their leases before the process exits. +// +// Skipping this does not lose data (every leased worker is restartable and the +// leases expire on their own), but it does mean the replacement instance waits +// out the full TTL before it can take over, so a rolling deploy pauses indexing +// for a minute instead of a second. +func (s *AppViewServer) waitForLeasedWorkers() { + if s.Leases == nil { + return + } + const leaseDrainTimeout = 5 * time.Second + slog.Info("Waiting for leased workers to release", "component", "leases", "timeout", leaseDrainTimeout) + s.Leases.Wait(leaseDrainTimeout) +} + +// startCleanupWorker runs the periodic expiry sweep under the cleanup lease. +func (s *AppViewServer) startCleanupWorker(ctx context.Context) { + const cleanupInterval = time.Hour + s.Leases.Go(ctx, db.LeaseCleanup, func(leaseCtx context.Context) error { + return db.RunPeriodicCleanup(leaseCtx, s.Database, s.SessionStore, cleanupInterval) + }) +} + // initializeJetstream initializes the Jetstream workers for real-time events and backfill. -func (s *AppViewServer) initializeJetstream() { +// +// Both run under leases. The consumer must be a singleton for correctness: +// StatsCache is per-process in-memory state whose aggregate is written to +// repository_stats as an absolute value, so two consumers each write a partial +// sum as though it were the whole truth. The webhook dispatcher hangs off the +// same processor, so a second consumer also doubles every delivery. Backfill is +// leased because it is expensive and re-reads every user's PDS. +func (s *AppViewServer) initializeJetstream(ctx context.Context) { jetstreamURLs := s.Config.Jetstream.URLs // Explicitly empty URLs disables Jetstream. The YAML config always @@ -939,15 +1010,21 @@ func (s *AppViewServer) initializeJetstream() { return } - go func() { + s.Leases.Go(ctx, db.LeaseJetstream, func(leaseCtx context.Context) error { worker := jetstream.NewWorker(s.Database, jetstreamURLs, 0) // Set webhook dispatcher on live worker (backfill skips dispatch) if s.WebhookDispatcher != nil { worker.Processor().SetWebhookDispatcher(s.WebhookDispatcher) } - worker.StartWithFailover(context.Background()) - }() - slog.Info("Jetstream real-time worker started", "component", "jetstream", "endpoints", len(jetstreamURLs)) + slog.Info("Jetstream real-time worker started", "component", "jetstream", "endpoints", len(jetstreamURLs)) + // Returns only when leaseCtx is done: either shutdown, or the lease was + // lost and this instance must stop consuming before the new holder + // starts. A fresh worker is built on each acquisition so its in-memory + // StatsCache is never carried across a gap in which another instance + // was the one indexing. + worker.StartWithFailover(leaseCtx) + return leaseCtx.Err() + }) if s.Config.Jetstream.BackfillEnabled { relayEndpoints := s.Config.Jetstream.RelayEndpoints @@ -958,38 +1035,52 @@ func (s *AppViewServer) initializeJetstream() { if err != nil { slog.Warn("Failed to create backfill worker", "component", "jetstream/backfill", "error", err) } else { - go func() { - startupDelay := 5 * time.Second - slog.Info("Waiting for services to be ready", "component", "jetstream/backfill", "startup_delay", startupDelay) - time.Sleep(startupDelay) - - slog.Info("Starting sync-based backfill", "component", "jetstream/backfill", "relay_endpoints", relayEndpoints) - if err := backfillWorker.Start(context.Background()); err != nil { - slog.Warn("Backfill finished with error", "component", "jetstream/backfill", "error", err) - } else { - slog.Info("Backfill completed successfully", "component", "jetstream/backfill") - } - }() - interval := s.Config.Jetstream.BackfillInterval - if interval > 0 { - go func() { - ticker := time.NewTicker(interval) - defer ticker.Stop() + // The startup run and the periodic schedule were two goroutines on + // context.Background(). They are one leased worker now, so a single + // instance backfills and a shutdown actually stops it mid-run + // instead of letting it finish against a closing database. + s.Leases.Go(ctx, db.LeaseBackfill, func(leaseCtx context.Context) error { + const startupDelay = 5 * time.Second + slog.Info("Waiting for services to be ready", "component", "jetstream/backfill", "startup_delay", startupDelay) + select { + case <-leaseCtx.Done(): + return leaseCtx.Err() + case <-time.After(startupDelay): + } - for range ticker.C { - slog.Info("Starting periodic backfill", "component", "jetstream/backfill", "interval", interval) - if err := backfillWorker.Start(context.Background()); err != nil { - slog.Warn("Periodic backfill finished with error", "component", "jetstream/backfill", "error", err) - } else { - slog.Info("Periodic backfill completed successfully", "component", "jetstream/backfill") - } + runBackfill := func(reason string) { + slog.Info("Starting backfill", "component", "jetstream/backfill", "reason", reason, "relay_endpoints", relayEndpoints) + if err := backfillWorker.Start(leaseCtx); err != nil { + slog.Warn("Backfill finished with error", "component", "jetstream/backfill", "reason", reason, "error", err) + } else { + slog.Info("Backfill completed successfully", "component", "jetstream/backfill", "reason", reason) } - }() + } + + runBackfill("startup") + + if interval <= 0 { + slog.Info("Periodic backfill disabled (interval=0), only startup backfill will run", "component", "jetstream/backfill") + // Hold the lease rather than returning: releasing it would + // let another instance acquire and run its own startup + // backfill, turning "once" into "once per instance". + <-leaseCtx.Done() + return leaseCtx.Err() + } + slog.Info("Periodic backfill scheduler started", "component", "jetstream/backfill", "interval", interval) - } else { - slog.Info("Periodic backfill disabled (interval=0), only startup backfill will run", "component", "jetstream/backfill") - } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-leaseCtx.Done(): + return leaseCtx.Err() + case <-ticker.C: + runBackfill("periodic") + } + } + }) } } }