mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-02 16:26:56 +00:00
Evan JarrettandClaude Opus 5
6a7ddb819b
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>
94 lines
3.1 KiB
Go
94 lines
3.1 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// InitializeDatabase initializes the libSQL database and session store.
|
|
// Returns: (read-write DB, read-only DB, session store)
|
|
func InitializeDatabase(dbPath string, cfg LibsqlConfig) (*sql.DB, *sql.DB, *SessionStore) {
|
|
// Ensure directory exists
|
|
dbDir := filepath.Dir(dbPath)
|
|
if err := os.MkdirAll(dbDir, 0700); err != nil {
|
|
slog.Warn("Failed to create UI database directory", "error", err)
|
|
return nil, nil, nil
|
|
}
|
|
|
|
// Initialize read-write database (for writes and auth operations)
|
|
database, err := InitDB(dbPath, cfg)
|
|
if err != nil {
|
|
slog.Warn("Failed to initialize UI database", "error", err)
|
|
return nil, nil, nil
|
|
}
|
|
|
|
// Open read-only connection for public queries (search, user pages, etc.)
|
|
// Uses ?mode=ro to prevent writes from public-facing handlers
|
|
roDSN := dbPath
|
|
if !strings.HasPrefix(dbPath, "file:") && !strings.HasPrefix(dbPath, ":memory:") {
|
|
roDSN = "file:" + dbPath
|
|
}
|
|
// Append ?mode=ro for read-only access
|
|
if strings.Contains(roDSN, "?") {
|
|
roDSN += "&mode=ro"
|
|
} else {
|
|
roDSN += "?mode=ro"
|
|
}
|
|
// Wrap with busyTimeoutConnector so every pooled read-only connection
|
|
// gets PRAGMA busy_timeout. Without this, reads return SQLITE_BUSY
|
|
// immediately when a write is in progress on the read-write connection
|
|
// (busy_timeout is per-connection, so a one-shot PRAGMA only configures
|
|
// whichever conn served it).
|
|
roBase, err := openLibsqlLocalConnector(roDSN)
|
|
if err != nil {
|
|
slog.Warn("Failed to open read-only database connector", "error", err)
|
|
return nil, nil, nil
|
|
}
|
|
readOnlyDB := sql.OpenDB(&busyTimeoutConnector{base: roBase, timeoutMs: 5000})
|
|
|
|
slog.Info("UI database initialized", "mode", "readonly", "path", dbPath)
|
|
|
|
// Create session store
|
|
sessionStore := NewSessionStore(database)
|
|
|
|
// 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)
|
|
}
|
|
}
|
|
}
|