diff --git a/pkg/auth/denial_counter_test.go b/pkg/auth/denial_counter_test.go new file mode 100644 index 0000000..e6eabdc --- /dev/null +++ b/pkg/auth/denial_counter_test.go @@ -0,0 +1,194 @@ +package auth + +import ( + "database/sql" + "path/filepath" + "sync" + "testing" + "time" + + "atcr.io/pkg/appview/db" +) + +// concurrentTestDB returns a file-backed database. +// +// It must not be ":memory:" like setupTestDB: go-libsql gives each connection to +// an in-memory DSN its own private database, so concurrent goroutines would each +// see a different (empty) one and the test would prove nothing. +func concurrentTestDB(t *testing.T) *sql.DB { + t.Helper() + testDB, err := db.InitDB(filepath.Join(t.TempDir(), "auth.db"), db.LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + t.Cleanup(func() { testDB.Close() }) + return testDB +} + +// TestCacheDenialConcurrentIncrementsAreNotLost is the regression test for the +// read-modify-write in cacheDenial. +// +// The old implementation read denial_count, incremented it in Go, and wrote the +// result back. Two overlapping denials for the same (hold, user) both read the +// same value and both wrote the same value, so one increment vanished. The +// visible effect is that the backoff ladder advances more slowly than +// configured, meaning a denied client keeps hammering the hold's PDS for longer +// than intended. +// +// This was already reachable across goroutines on one instance and becomes +// routine with several instances behind a load balancer. +func TestCacheDenialConcurrentIncrementsAreNotLost(t *testing.T) { + testDB := concurrentTestDB(t) + remote := NewRemoteHoldAuthorizerWithBackoffs( + testDB, false, + time.Hour, // firstDenialBackoff + time.Hour, // cleanupInterval + time.Hour, // cleanupGracePeriod + []time.Duration{time.Hour}, + ).(*RemoteHoldAuthorizer) + defer close(remote.stopCleanup) + + const ( + holdDID = "did:web:hold01.atcr.io" + userDID = "did:plc:user1" + ) + + // Two calls to get past the in-memory-only first denial and create the row. + if err := remote.cacheDenial(holdDID, userDID); err != nil { + t.Fatalf("first denial: %v", err) + } + if err := remote.cacheDenial(holdDID, userDID); err != nil { + t.Fatalf("second denial: %v", err) + } + + before := denialCount(t, testDB, holdDID, userDID) + if before != 1 { + t.Fatalf("expected the first persisted denial to be count 1, got %d", before) + } + + const concurrent = 20 + var wg sync.WaitGroup + errs := make(chan error, concurrent) + start := make(chan struct{}) + for range concurrent { + wg.Go(func() { + <-start + if err := remote.cacheDenial(holdDID, userDID); err != nil { + errs <- err + } + }) + } + close(start) + wg.Wait() + close(errs) + + for err := range errs { + t.Errorf("cacheDenial: %v", err) + } + + got := denialCount(t, testDB, holdDID, userDID) + want := before + concurrent + if got != want { + t.Errorf("denial_count = %d, want %d: %d increments were lost to the read-modify-write", + got, want, want-got) + } +} + +// TestCacheDenialBackoffMatchesLadder checks that next_retry_at, now computed in +// SQL from the count the same statement produces, still lands where +// getBackoffDuration says it should. +func TestCacheDenialBackoffMatchesLadder(t *testing.T) { + testDB := concurrentTestDB(t) + ladder := []time.Duration{2 * time.Second, 30 * time.Second, 5 * time.Minute} + remote := NewRemoteHoldAuthorizerWithBackoffs( + testDB, false, time.Hour, time.Hour, time.Hour, ladder, + ).(*RemoteHoldAuthorizer) + defer close(remote.stopCleanup) + + const ( + holdDID = "did:web:hold01.atcr.io" + userDID = "did:plc:user1" + ) + + // First denial is in-memory only and creates no row. + if err := remote.cacheDenial(holdDID, userDID); err != nil { + t.Fatalf("first denial: %v", err) + } + + // Each subsequent denial advances one rung, clamping at the last. + wantByCount := map[int]time.Duration{ + 1: ladder[0], + 2: ladder[1], + 3: ladder[2], + 4: ladder[2], // clamped + 5: ladder[2], + } + + for count := 1; count <= 5; count++ { + issued := time.Now() + if err := remote.cacheDenial(holdDID, userDID); err != nil { + t.Fatalf("denial %d: %v", count, err) + } + + var gotCount int + var nextRetry time.Time + err := testDB.QueryRow( + `SELECT denial_count, next_retry_at FROM hold_crew_denials WHERE hold_did = ? AND user_did = ?`, + holdDID, userDID, + ).Scan(&gotCount, &nextRetry) + if err != nil { + t.Fatalf("read denial %d: %v", count, err) + } + if gotCount != count { + t.Fatalf("denial_count = %d, want %d", gotCount, count) + } + + want := wantByCount[count] + actual := nextRetry.Sub(issued) + // Generous tolerance: the value is SQLite's clock, not Go's, and the + // column stores milliseconds. + if actual < want-2*time.Second || actual > want+2*time.Second { + t.Errorf("count %d: next_retry_at is %v out, want about %v", count, actual, want) + } + } +} + +// TestCacheDenialBlocksAfterPersisting ties the new write back to the read path: +// a persisted denial must actually block. +func TestCacheDenialBlocksAfterPersisting(t *testing.T) { + testDB := concurrentTestDB(t) + remote := NewRemoteHoldAuthorizerWithBackoffs( + testDB, false, time.Hour, time.Hour, time.Hour, + []time.Duration{time.Hour}, + ).(*RemoteHoldAuthorizer) + defer close(remote.stopCleanup) + + const ( + holdDID = "did:web:hold01.atcr.io" + userDID = "did:plc:user1" + ) + + _ = remote.cacheDenial(holdDID, userDID) // in-memory + _ = remote.cacheDenial(holdDID, userDID) // persisted + + blocked, err := remote.isBlockedByDenialBackoff(holdDID, userDID) + if err != nil { + t.Fatalf("isBlockedByDenialBackoff: %v", err) + } + if !blocked { + t.Error("expected the user to be blocked by the persisted backoff") + } +} + +func denialCount(t *testing.T, database *sql.DB, holdDID, userDID string) int { + t.Helper() + var n int + err := database.QueryRow( + `SELECT denial_count FROM hold_crew_denials WHERE hold_did = ? AND user_did = ?`, + holdDID, userDID, + ).Scan(&n) + if err != nil { + t.Fatalf("read denial_count: %v", err) + } + return n +} diff --git a/pkg/auth/hold_remote.go b/pkg/auth/hold_remote.go index 06d2afa..65cc4ab 100644 --- a/pkg/auth/hold_remote.go +++ b/pkg/auth/hold_remote.go @@ -4,11 +4,14 @@ import ( "context" "database/sql" "encoding/json" + "errors" "fmt" "io" "log/slog" "net/http" "net/url" + "strconv" + "strings" "sync" "time" @@ -572,74 +575,151 @@ func (a *RemoteHoldAuthorizer) isBlockedByDenialBackoff(holdDID, userDID string) return false, nil } -// cacheDenial stores or updates a denial with exponential backoff +// sqliteNow is SQLite's clock rendered as RFC 3339 with milliseconds, matching +// what the driver writes for a Go time.Time closely enough that both parse back +// into time.Time correctly. +// +// datetime() is not usable here because it truncates to whole seconds, and the +// backoff durations can be sub-second in tests. +const sqliteNow = `strftime('%Y-%m-%dT%H:%M:%fZ','now')` + +// sqliteNowPlus returns the same, offset by a SQLite time modifier expression. +func sqliteNowPlus(modifier string) string { + return `strftime('%Y-%m-%dT%H:%M:%fZ','now', ` + modifier + `)` +} + +// backoffSecondsCaseSQL builds a CASE expression mapping the NEW denial count to +// its backoff in seconds, mirroring getBackoffDuration. +// +// getBackoffDuration indexes the ladder at newCount-1, clamped to the last +// entry, so entry i applies when newCount == i+1 and the final entry covers +// everything beyond. +// +// The values are durations from configuration, formatted as numbers, so there is +// no injection surface here: nothing from a request reaches this string. +func (a *RemoteHoldAuthorizer) backoffSecondsCaseSQL() string { + backoffs := a.dbBackoffDurations + if len(backoffs) == 0 { + return "0" + } + + secs := func(d time.Duration) string { + return strconv.FormatFloat(d.Seconds(), 'f', 3, 64) + } + + // A single-rung ladder applies to every count, and "CASE ELSE x END" with no + // WHEN arm is a syntax error, so emit the bare number. + if len(backoffs) == 1 { + return secs(backoffs[0]) + } + + var sb strings.Builder + sb.WriteString("CASE") + for i := range len(backoffs) - 1 { + fmt.Fprintf(&sb, " WHEN hold_crew_denials.denial_count + 1 <= %d THEN %s", i+1, secs(backoffs[i])) + } + fmt.Fprintf(&sb, " ELSE %s END", secs(backoffs[len(backoffs)-1])) + return sb.String() +} + +// cacheDenial stores or updates a denial with exponential backoff. // First denial: in-memory only (configurable backoff, default 10s) // Second+ denial: database with exponential backoff (configurable, default 1m/5m/15m/1h) +// +// The database write is a single atomic statement. It used to be a SELECT of +// denial_count, an increment in Go, and an upsert of the computed value, which +// loses increments when two requests for the same (hold, user) overlap. That was +// already possible across goroutines and becomes routine once more than one +// AppView instance serves traffic, and its effect is that the backoff escalates +// more slowly than configured, so a denied client keeps hammering the hold. +// +// next_retry_at is therefore computed in SQL too, from the count the same +// statement is producing, rather than in Go from a count that may already be +// stale by the time the write lands. func (a *RemoteHoldAuthorizer) cacheDenial(holdDID, userDID string) error { key := fmt.Sprintf("%s:%s", holdDID, userDID) // Check if this is a first denial (not in memory, not in DB) _, inMemory := a.recentDenials.Load(key) - var denialCount int - query := `SELECT denial_count FROM hold_crew_denials WHERE hold_did = ? AND user_did = ?` - err := a.db.QueryRow(query, holdDID, userDID).Scan(&denialCount) - - inDB := err != sql.ErrNoRows - if err != nil && err != sql.ErrNoRows { - return err + if !inMemory { + var existing int + err := a.db.QueryRow( + `SELECT denial_count FROM hold_crew_denials WHERE hold_did = ? AND user_did = ?`, + holdDID, userDID, + ).Scan(&existing) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if errors.Is(err, sql.ErrNoRows) { + // First denial: store only in memory with configurable backoff. + // This read only picks a branch; it is no longer the source of the + // increment, so a stale answer costs at most one skipped or one + // extra database write, never a lost count. + now := time.Now() + a.recentDenials.Store(key, denialEntry{timestamp: now}) + slog.Info("Cached first crew denial (in-memory)", + "holdDID", holdDID, + "userDID", userDID, + "denial_count", 1, + "backoff_type", "in_memory", + "backoff_duration", a.firstDenialBackoff, + "retry_after", now.Add(a.firstDenialBackoff)) + return nil + } } - // If not in memory and not in DB, this is the first denial - if !inMemory && !inDB { - // First denial: store only in memory with configurable backoff - now := time.Now() - a.recentDenials.Store(key, denialEntry{timestamp: now}) - slog.Info("Cached first crew denial (in-memory)", - "holdDID", holdDID, - "userDID", userDID, - "denial_count", 1, - "backoff_type", "in_memory", - "backoff_duration", a.firstDenialBackoff, - "retry_after", now.Add(a.firstDenialBackoff)) - return nil - } - - // Second+ denial: persist to database with exponential backoff - denialCount++ - backoff := a.getBackoffDuration(denialCount) - now := time.Now() - nextRetry := now.Add(backoff) - - // Upsert denial record + // Second+ denial: one statement, so concurrent denials for the same + // (hold, user) each add exactly one. upsertQuery := ` INSERT INTO hold_crew_denials (hold_did, user_did, denial_count, next_retry_at, last_denied_at) - VALUES (?, ?, ?, ?, ?) + VALUES (?, ?, 1, ` + sqliteNowPlus(`'+' || `+firstBackoffSeconds(a)+` || ' seconds'`) + `, ` + sqliteNow + `) ON CONFLICT(hold_did, user_did) DO UPDATE SET - denial_count = excluded.denial_count, - next_retry_at = excluded.next_retry_at, - last_denied_at = excluded.last_denied_at + denial_count = hold_crew_denials.denial_count + 1, + next_retry_at = ` + sqliteNowPlus(`'+' || (`+a.backoffSecondsCaseSQL()+`) || ' seconds'`) + `, + last_denied_at = ` + sqliteNow + ` ` - _, err = a.db.Exec(upsertQuery, holdDID, userDID, denialCount, nextRetry, now) - if err != nil { + if _, err := a.db.Exec(upsertQuery, holdDID, userDID); err != nil { return err } // Remove from in-memory cache since we're now tracking in DB a.recentDenials.Delete(key) + // Read back purely for the log line. Denial diagnostics are the reason this + // logging exists, so it is worth a cheap extra read, but a failure here must + // not fail the denial itself. + var denialCount int + var nextRetry time.Time + if err := a.db.QueryRow( + `SELECT denial_count, next_retry_at FROM hold_crew_denials WHERE hold_did = ? AND user_did = ?`, + holdDID, userDID, + ).Scan(&denialCount, &nextRetry); err != nil { + slog.Debug("Could not read back denial state for logging", "error", err) + return nil + } + slog.Info("Cached crew denial with exponential backoff", "holdDID", holdDID, "userDID", userDID, "denial_count", denialCount, "backoff_type", "database", - "backoff_duration", backoff, + "backoff_duration", a.getBackoffDuration(denialCount), "next_retry_at", nextRetry) return nil } +// firstBackoffSeconds renders the backoff for a freshly inserted denial row +// (count 1), which getBackoffDuration maps to the first entry in the ladder. +func firstBackoffSeconds(a *RemoteHoldAuthorizer) string { + if len(a.dbBackoffDurations) == 0 { + return "0" + } + return strconv.FormatFloat(a.dbBackoffDurations[0].Seconds(), 'f', 3, 64) +} + // getBackoffDuration returns the backoff duration based on denial count // Note: First denial is in-memory only and not tracked by this function // This function handles second+ denials using configurable durations