appview: run background workers under a lease

The Jetstream consumer, backfill, labeler subscriber, cleanup sweep and billing
tier refresh all started unconditionally in every process. That is correct for
one instance and wrong for two.

The consumer is the case with teeth. StatsCache is per-process in-memory state,
and the aggregate it produces is written to repository_stats as an absolute
value rather than an increment, so two consumers each hold a partial view of the
holds and each write their partial sum as though it were the whole truth,
overwriting one another indefinitely. The webhook dispatcher hangs off the same
processor, so a second consumer also doubles every delivery.

Each now runs under a named lease, so exactly one instance runs it and a
replacement takes over when that instance goes away. The health worker is
deliberately not leased: it refreshes a cache each instance needs locally, so
running it everywhere is correct.

Two structural changes came with it. The cleanup loop moved out of
InitializeDatabase, where it was a bare goroutine with no way to reach the lease
manager, into RunPeriodicCleanup called from the server. And backfill's startup
run and periodic schedule became one leased worker instead of two goroutines on
context.Background(), so shutdown actually stops a backfill in flight rather
than letting it run on against a closing database. With interval=0 that worker
holds its lease instead of returning, since releasing would let another instance
acquire and run its own startup backfill, turning "once" into "once per
instance".

Verified with two instances against one database: exactly one acquired, the
other contended without starting a worker; SIGTERM handed over in 13ms via the
release, SIGKILL handed over in ~12s via TTL expiry.

That first number only holds because of Manager.Go and Manager.Wait, which this
commit adds. The first cut used `go m.Run(...)` and cancelled the worker context
during shutdown without waiting, so the process exited before the release landed
and the lease survived to its TTL — a rolling deploy would have paused indexing
for a minute rather than a second. Nothing in the unit tests caught it; the
two-instance run did. TestWaitBlocksUntilLeaseReleased covers it now.

leases.enabled defaults to true. A single instance is unaffected, since it
always wins its own leases, while an operator who scales out without reading the
docs still gets correct behavior instead of silent stats corruption.

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

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.

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

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",

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)
+		}
+	}
+}

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.
 //

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) {

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")
+					}
+				}
+			})
 		}
 	}
 }