From dd89881a81cdf668ee653c3f7d9fc9569d2bcfad Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 16:38:51 +0300 Subject: [PATCH 1/9] feat(auth): declare the otp.prune task types and its config (RUK-296) Adds otp.prune and otp.prune.cron plus ProcessorTaskPayloadOTPPrune, and the task_processor.otp_prune config block with validateOTPRetention. The two task-type registries are guarded differently and both entries are required: the runtime startup guard reads ActiveProcessorTaskTypes, while the drift test's length assertion reads allDeclaredTaskTypes. The validator checks only for a negative retention, deliberately. A comparison against the OTP TTL was considered and dropped: a code becomes sweep-eligible only once its expires_at is already past, so no positive retention can reach a live row however small -- and services/otp imports this package, so calling otp.TTL from config would not compile anyway. Co-Authored-By: Claude Opus 5 --- internal/config/app_config.go | 44 +++++++++++++ internal/config/app_config_otp_prune_test.go | 65 +++++++++++++++++++ internal/entity/goque_processor_owner_test.go | 2 + internal/entity/goque_processors_task.go | 24 +++++++ internal/entity/goque_processors_task_test.go | 27 ++++++++ 5 files changed, 162 insertions(+) create mode 100644 internal/config/app_config_otp_prune_test.go diff --git a/internal/config/app_config.go b/internal/config/app_config.go index 78f20750..f8e96557 100644 --- a/internal/config/app_config.go +++ b/internal/config/app_config.go @@ -322,6 +322,7 @@ type TaskProcessorConfig struct { AuditPrune TaskProcessorAuditPruneConfig `mapstructure:"audit_prune"` InvitationRotate TaskProcessorInvitationRotateConfig `mapstructure:"invitation_rotate"` InvitationPrune TaskProcessorInvitationPruneConfig `mapstructure:"invitation_prune"` + OTPPrune TaskProcessorOTPPruneConfig `mapstructure:"otp_prune"` } // CryptoConfig addresses the master keys (KEKs) that wrap the data-encryption @@ -463,6 +464,24 @@ type TaskProcessorInvitationPruneConfig struct { BatchLimit int64 `mapstructure:"batch_limit"` } +// TaskProcessorOTPPruneConfig tunes the one-time-code retention sweep that +// deletes spent OTP credentials (see services/otp.Service.Prune). +type TaskProcessorOTPPruneConfig struct { + // CronSpec is the 5-field schedule for the producer job. Unlike the sibling + // sweeps this one has a code-side fallback, so leaving it empty degrades to + // the default schedule rather than aborting startup. The task is + // day-bucketed, so firing more often than daily still yields one prune a day. + CronSpec string `mapstructure:"cron_spec"` + // Retention is the age threshold: an OTP whose expires_at is older than + // now-Retention is deleted. Short by design (24h), because the row holds a + // code digest and a session nonce and the code itself lives only minutes. + // Worst-case residency is Retention plus one cron period. + Retention time.Duration `mapstructure:"retention"` + // BatchLimit bounds how many rows one DELETE statement removes; the sweep loops + // batches until the table is drained for the cutoff. + BatchLimit int64 `mapstructure:"batch_limit"` +} + type JWTVerifierConfig struct { // This struct is shared by two verifiers that use DIFFERENT issuer fields. // Set the one your consumer reads; validateIssuerConfig enforces both at @@ -633,6 +652,10 @@ func initConfig(appName string) *AppConfig { log.Panicf("invalid config for service %s: %s", appName, err) } + if err := cfg.validateOTPRetention(); err != nil { + log.Panicf("invalid config for service %s: %s", appName, err) + } + if err := cfg.validateValkeyConfig(); err != nil { log.Panicf("invalid config for service %s: %s", appName, err) } @@ -851,6 +874,27 @@ func (c *AppConfig) validateInvitationRetention() error { return nil } +// validateOTPRetention rejects a negative otp-prune retention at startup, for +// the same reason as its invitation twin: a negative value would push the prune +// cutoff into the future, and it is always an operator typo rather than an +// intent. The service clamps it defensively too — that clamp, not this check, is +// what actually keeps a live code from being deleted — but a bad value in config +// should still be loud. Zero is allowed and means "unset". +// +// Deliberately no comparison against the OTP TTL. A code is only eligible once +// its expires_at is already past, so every positive retention is safe however +// small, and reaching otp.TTL from here would be an import cycle anyway +// (services/otp imports this package). +func (c *AppConfig) validateOTPRetention() error { + if c.TaskProcessor.OTPPrune.Retention < 0 { + return fmt.Errorf( + "task_processor.otp_prune.retention must not be negative, got %s", + c.TaskProcessor.OTPPrune.Retention, + ) + } + return nil +} + // validateIssuerConfig rejects a verifier config that would silently stop // checking the token issuer. // diff --git a/internal/config/app_config_otp_prune_test.go b/internal/config/app_config_otp_prune_test.go new file mode 100644 index 00000000..a58d1aae --- /dev/null +++ b/internal/config/app_config_otp_prune_test.go @@ -0,0 +1,65 @@ +package config + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The validator is exercised directly rather than through LoadAppConfig: the +// initConfig chain reports a validation failure with log.Panicf, which would +// take the test binary down instead of failing an assertion. +func TestValidateOTPRetention(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + retention time.Duration + wantErr bool + }{ + { + name: "negative is rejected", + retention: -time.Hour, + wantErr: true, + }, + { + // Zero means "unset": the service applies its own default, so it must + // not be treated as a misconfiguration. + name: "zero is accepted as unset", + retention: 0, + wantErr: false, + }, + { + name: "shipped default is accepted", + retention: 24 * time.Hour, + wantErr: false, + }, + { + // A retention far below the code lifetime is a poor forensic window + // but not a safety problem: a row is only eligible once expires_at is + // already in the past, so no live code can be reached. Accepting it + // keeps the validator a mirror of its invitation sibling. + name: "small positive is accepted", + retention: time.Minute, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cfg := &AppConfig{} + cfg.TaskProcessor.OTPPrune.Retention = tt.retention + + err := cfg.validateOTPRetention() + if tt.wantErr { + require.Error(t, err) + require.Contains(t, err.Error(), "task_processor.otp_prune.retention") + return + } + require.NoError(t, err) + }) + } +} diff --git a/internal/entity/goque_processor_owner_test.go b/internal/entity/goque_processor_owner_test.go index add4f445..4a2c7732 100644 --- a/internal/entity/goque_processor_owner_test.go +++ b/internal/entity/goque_processor_owner_test.go @@ -26,6 +26,8 @@ var allDeclaredTaskTypes = []string{ ProcessorTaskInvitationPrune, ProcessorTaskInvitationPruneCron, ProcessorTaskOTPEmailSend, + ProcessorTaskOTPPrune, + ProcessorTaskOTPPruneCron, } // disabledTaskTypes is every declared type whose processor is intentionally not diff --git a/internal/entity/goque_processors_task.go b/internal/entity/goque_processors_task.go index da8cf88c..93468860 100644 --- a/internal/entity/goque_processors_task.go +++ b/internal/entity/goque_processors_task.go @@ -81,6 +81,18 @@ const ( // generic sender processor could not decode it. The body does not exist at // enqueue time -- it is rendered by this type's own processor. ProcessorTaskOTPEmailSend = "otp.email" + // ProcessorTaskOTPPrune is the goque task type produced by the OTP-retention + // periodic job. Its processor deletes auth_credentials rows with kind='otp' + // whose expires_at is older than the retention window, in bounded batches. + // + // Age is measured on expires_at, not created_at, because + // auth_credentials_otp_expiry_idx is the only index on the table and + // migrations were out of scope. A consumed code carries an expiry too and + // ages out through the same sweep, so consumed_at never enters the predicate. + // password rows are never eligible. The payload carries the retention window + // and batch limit (from config). + ProcessorTaskOTPPrune = "otp.prune" + ProcessorTaskOTPPruneCron = "otp.prune.cron" ) // ActiveProcessorTaskTypes is the set of goque task types the process must @@ -108,6 +120,8 @@ var ActiveProcessorTaskTypes = map[string]struct{}{ ProcessorTaskInvitationPrune: {}, ProcessorTaskInvitationPruneCron: {}, ProcessorTaskOTPEmailSend: {}, + ProcessorTaskOTPPrune: {}, + ProcessorTaskOTPPruneCron: {}, } // ExpectedProcessorTaskTypes returns the exact task-type set the process must @@ -214,6 +228,16 @@ type ProcessorTaskPayloadInvitationPrune struct { BatchLimit int64 `json:"batch_limit"` } +// ProcessorTaskPayloadOTPPrune is the payload of an OTP-retention sweep task. +// Same shape and same reason as the invitation one: the cron job stamps the +// tunables from config so the processor stays config-free. Retention is the age +// past expires_at at which a one-time code is deleted, BatchLimit bounds how +// many rows one DELETE removes. +type ProcessorTaskPayloadOTPPrune struct { + Retention time.Duration `json:"retention"` + BatchLimit int64 `json:"batch_limit"` +} + // ProcessorTaskPayloadAuditWrite is the payload of an audit-write task. // It is the rendered, point-in-time snapshot of one audit event: // the publisher fills every persisted field at dispatch time and the processor diff --git a/internal/entity/goque_processors_task_test.go b/internal/entity/goque_processors_task_test.go index 60da1c34..e460369f 100644 --- a/internal/entity/goque_processors_task_test.go +++ b/internal/entity/goque_processors_task_test.go @@ -111,3 +111,30 @@ func TestProcessorTaskPayloadAuditWrite_ToAuditEntry(t *testing.T) { require.Equal(t, payload.Details, entry.Details) require.Equal(t, payload.Metadata, entry.Metadata) } + +func TestOTPPruneTaskTypes_RegisteredInGuard(t *testing.T) { + // Same contract as the invitation pair above: both the task and its .cron + // producer must sit in the active set, or NewTaskProcessors registers a + // processor the guard does not expect and startup fails. + for _, taskType := range []string{ + ProcessorTaskOTPPrune, + ProcessorTaskOTPPruneCron, + } { + _, ok := ActiveProcessorTaskTypes[taskType] + require.Truef(t, ok, "task type %q must be in ActiveProcessorTaskTypes", taskType) + } +} + +func TestProcessorTaskPayloadOTPPrune_JSONRoundTrip(t *testing.T) { + want := ProcessorTaskPayloadOTPPrune{ + Retention: 24 * time.Hour, + BatchLimit: 1000, + } + + raw, err := json.Marshal(want) + require.NoError(t, err) + + var got ProcessorTaskPayloadOTPPrune + require.NoError(t, json.Unmarshal(raw, &got)) + require.Equal(t, want, got) +} From df02ccc7ca114c3c4cd02ea305f50b9550e98d76 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 16:39:07 +0300 Subject: [PATCH 2/9] feat(auth): delete spent one-time codes in the credentials store (RUK-296) PruneOTPExpiredBefore is one batch-bounded DELETE of kind='otp' rows past expires_at, ordered by the indexed column so the partial index serves the scan directly rather than through a backward one. Age is measured on expires_at rather than created_at because auth_credentials_otp_expiry_idx is the only index on the table and migrations were out of scope; the two differ by exactly the TTL for a one-time code. consumed_at never enters the predicate -- a consumed code carries an expiry too and ages out through the same sweep. The kind='otp' conjunct is load-bearing twice and must not be simplified away on the grounds that password rows have a NULL expiry: no CHECK ties kind to a NULL expires_at, so that is an observation about today's write paths rather than an invariant, and the literal is also what lets Postgres use the PARTIAL index at all. Two fixtures are new because the existing helpers could not express the rows these tests need -- makeOTP hardcodes now+10m, and makePassword sets no expiry. makePasswordExpiringAt in particular builds a shape no write path produces today, and it is the only fixture that can prove the kind guard does anything: a NULL-expiry password row survives whether or not the guard exists. Every cutoff here stays in the past by design. The DELETE is bounded by expires_at and limit alone -- never by user_id -- so under `-p 2` a sweeping cutoff would eat fixtures belonging to other packages, and the bad outcome is not a flake but a silent green in services/otp's TestVerify_ExpiredCodeIsRetired, whose "row is gone" assertion an eaten fixture also satisfies. Mutation-verified: dropping the kind conjunct fails SparesPasswordWithExpiry and only that test; replacing the cutoff comparison with IS NOT NULL fails SparesLiveCode at its 1-minute leg. Co-Authored-By: Claude Opus 5 --- .../storages/authcredentials/main_test.go | 54 +++++ internal/storages/authcredentials/prune.go | 70 ++++++ .../storages/authcredentials/prune_test.go | 211 ++++++++++++++++++ 3 files changed, 335 insertions(+) create mode 100644 internal/storages/authcredentials/prune.go create mode 100644 internal/storages/authcredentials/prune_test.go diff --git a/internal/storages/authcredentials/main_test.go b/internal/storages/authcredentials/main_test.go index 1e0a7657..ffd30239 100644 --- a/internal/storages/authcredentials/main_test.go +++ b/internal/storages/authcredentials/main_test.go @@ -73,6 +73,60 @@ func makeOTP(ctx context.Context, t *testing.T, userID uuid.UUID) *entity.AuthCr return cred } +// makeOTPExpiringAt inserts a one-time code with a caller-chosen expiry, so a +// test can place a row on either side of a prune cutoff. Create inserts +// expires_at straight from the entity, so no back-dating UPDATE is needed here — +// unlike user_invitations, whose created_at is stamped server-side. +func makeOTPExpiringAt( + ctx context.Context, + t *testing.T, + userID uuid.UUID, + expiresAt time.Time, +) *entity.AuthCredential { + t.Helper() + + nonce := uuid.NewString() + + cred, err := store.Create(ctx, &entity.AuthCredential{ + UserID: userID, + Kind: entity.AuthCredentialKindOTP, + SecretHash: uuid.NewString(), + ExpiresAt: &expiresAt, + SessionNonce: &nonce, + }) + require.NoError(t, err) + require.NotNil(t, cred) + + return cred +} + +// makePasswordExpiringAt inserts a password credential that carries an expiry. +// +// Nothing in the schema forbids one — no CHECK ties kind='password' to a NULL +// expires_at — and that is exactly why this fixture exists: it is the only row +// shape that proves the prune's kind guard does real work. A password row with a +// NULL expiry cannot prove it, because NULL < cutoff is never true and such a row +// survives whether or not the guard is there. +func makePasswordExpiringAt( + ctx context.Context, + t *testing.T, + userID uuid.UUID, + expiresAt time.Time, +) *entity.AuthCredential { + t.Helper() + + cred, err := store.Create(ctx, &entity.AuthCredential{ + UserID: userID, + Kind: entity.AuthCredentialKindPassword, + SecretHash: "$argon2id$v=19$m=65536,t=3,p=4$" + uuid.NewString(), + ExpiresAt: &expiresAt, + }) + require.NoError(t, err) + require.NotNil(t, cred) + + return cred +} + // makePassword inserts a password credential for the user. func makePassword(ctx context.Context, t *testing.T, userID uuid.UUID) *entity.AuthCredential { t.Helper() diff --git a/internal/storages/authcredentials/prune.go b/internal/storages/authcredentials/prune.go new file mode 100644 index 00000000..82ba4f49 --- /dev/null +++ b/internal/storages/authcredentials/prune.go @@ -0,0 +1,70 @@ +package authcredentials + +import ( + "context" + "time" + + "github.com/go-jet/jet/v2/postgres" + "github.com/ruko1202/xlog" + + "github.com/ruko1202/maintmode/internal/entity" + "github.com/ruko1202/maintmode/internal/pkg/generated/maintmode/public/table" +) + +// PruneOTPExpiredBefore deletes up to limit one-time codes whose expires_at is +// strictly before cutoff, returning how many it removed. It is the single batch +// behind the retention sweep: the caller loops until a batch deletes fewer than +// limit. +// +// Age is measured on expires_at, not created_at. auth_credentials_otp_expiry_idx +// is the only index on this table and it is on expires_at; for a one-time code +// the two columns differ by exactly the TTL, so nothing is lost by preferring +// the indexed one. +// +// consumed_at is deliberately absent from the predicate. A consumed code still +// carries an expiry and ages out through this same sweep, so no second branch +// (and no second tunable) is needed to collect it. +// +// The `kind = 'otp'` conjunct is load-bearing twice, and must not be "simplified" +// away on the grounds that password rows have a NULL expiry: +// +// - Correctness. Nothing in the schema ties kind='password' to a NULL +// expires_at — there is no such CHECK — so the NULL barrier is an observation +// about today's write paths, not an invariant. A password row that ever +// gained an expiry would be deleted by a predicate resting on NULL alone. +// - The query plan. The index is PARTIAL (WHERE kind = 'otp'), and Postgres can +// only use it if the query's predicate implies the index's. Drop this literal +// and the sweep silently degrades to a sequential scan plus sort. No test can +// observe that, which is why it is written down here. +// +// ORDER BY expires_at ASC matches the index's own direction, so it is served +// directly rather than by a backward scan. +// +// Postgres has no DELETE ... LIMIT, so the batch is bounded by an id-subquery: +// pick the oldest `limit` eligible ids, then delete exactly those. Bounding each +// batch keeps the per-statement lock footprint small on a table that takes a row +// per sign-in attempt. +func (s *Store) PruneOTPExpiredBefore(ctx context.Context, cutoff time.Time, limit int64) (int64, error) { + ctx, span := xlog.WithOperationSpan(ctx, "store.AuthCredentials.PruneOTPExpiredBefore") + defer span.End() + + expired := table.AuthCredentials. + SELECT(table.AuthCredentials.ID). + WHERE( + table.AuthCredentials.Kind.EQ(postgres.String(string(entity.AuthCredentialKindOTP))). + AND(table.AuthCredentials.ExpiresAt.LT(postgres.TimestampzT(cutoff))), + ). + ORDER_BY(table.AuthCredentials.ExpiresAt.ASC()). + LIMIT(limit) + + stmt := table.AuthCredentials. + DELETE(). + WHERE(table.AuthCredentials.ID.IN(expired)) + + res, err := stmt.ExecContext(ctx, s.db.Executor(ctx)) + if err != nil { + return 0, err + } + + return res.RowsAffected() +} diff --git a/internal/storages/authcredentials/prune_test.go b/internal/storages/authcredentials/prune_test.go new file mode 100644 index 00000000..8f28ac3c --- /dev/null +++ b/internal/storages/authcredentials/prune_test.go @@ -0,0 +1,211 @@ +package authcredentials + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +// Two separate rules govern this file, and only the first is about assertions. +// +// 1. Every assertion counts rows for one freshly created user. The suite runs +// with -count 2 against a shared database, so a table-wide count would race +// whatever else is running. +// +// 2. Every cutoff passed to the sweep stays in the PAST, and close to it. This +// one is easy to miss because it is not about what a test observes but about +// what its DELETE reaches: the statement is bounded by expires_at and limit +// alone -- never by user_id -- so a distant cutoff deletes rows belonging to +// whatever else shares the database, including other packages running +// concurrently under `-p 2`. Scoping the assertions does not make a sweeping +// cutoff safe. +func credentialExists(ctx context.Context, t *testing.T, id uuid.UUID) bool { + t.Helper() + + var n int + require.NoError(t, db.GetContext(ctx, &n, + "SELECT count(*) FROM auth_credentials WHERE id = $1", id)) + + return n > 0 +} + +func countForUser(ctx context.Context, t *testing.T, userID uuid.UUID) int { + t.Helper() + + var n int + require.NoError(t, db.GetContext(ctx, &n, + "SELECT count(*) FROM auth_credentials WHERE user_id = $1", userID)) + + return n +} + +// TestPruneOTPExpiredBefore_DeletesOnlyRowsPastCutoff covers the core age +// boundary: a code expired before the cutoff goes, one expiring after it stays. +func TestPruneOTPExpiredBefore_DeletesOnlyRowsPastCutoff(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + // Two users, because one live OTP per user is a partial-unique invariant. + oldUser := makeUser(ctx, t) + freshUser := makeUser(ctx, t) + + old := makeOTPExpiringAt(ctx, t, oldUser.ID, now.Add(-48*time.Hour)) + fresh := makeOTPExpiringAt(ctx, t, freshUser.ID, now.Add(10*time.Minute)) + + deleted, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + require.GreaterOrEqual(t, deleted, int64(1)) + + require.False(t, credentialExists(ctx, t, old.ID), "expired code must be deleted") + require.True(t, credentialExists(ctx, t, fresh.ID), "unexpired code must survive") +} + +// TestPruneOTPExpiredBefore_SparesLiveCode is the security criterion, and the +// reason it stands apart from the boundary test above. +// +// The per-code attempt ceiling is enforced by the row's continued existence: +// claimSlot refuses to issue a new code while it finds a live row with the +// attempts exhausted. Delete such a row and the next request takes the +// first-request branch and hands back a fresh code with a fresh counter, turning +// "five attempts per code" into "five attempts per code, unlimited codes". +// +// The property is checked at the near edge of the permitted range, not only at +// the 24h default: a 1-minute retention is the smallest cutoff config allows, +// and if a future-dated row survives that, it survives every larger one. The +// attempts counter is set to the ceiling so the fixture is exactly the row the +// bypass would need. +func TestPruneOTPExpiredBefore_SparesLiveCode(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + user := makeUser(ctx, t) + live := makeOTPExpiringAt(ctx, t, user.ID, now.Add(5*time.Minute)) + + // Exhaust the guess ceiling: this is the state the barrier exists to hold. + claimed, err := store.ClaimOTPAttempt(ctx, live.ID, 1) + require.NoError(t, err) + require.True(t, claimed) + + for _, retention := range []time.Duration{time.Minute, 24 * time.Hour} { + _, err := store.PruneOTPExpiredBefore(ctx, now.Add(-retention), 100) + require.NoError(t, err) + require.True(t, credentialExists(ctx, t, live.ID), + "a live code must survive a sweep at retention %s", retention) + } +} + +// TestPruneOTPExpiredBefore_SparesPasswordWithNullExpiry is the NULL barrier: +// a password row has no expiry, and NULL < cutoff is never true. +// +// The cutoff is in the PAST, deliberately. A far-future cutoff would demonstrate +// the same property, but it would also make every OTP row in the database +// eligible -- and the DELETE's reach is not scoped by user_id even though this +// file's assertions are. Under `-p 2` another package's DB-backed suite can be +// running against the same table, so a sweeping cutoff would delete fixtures out +// from under it: a flaky red in services/otp's claim tests, and worse, a silent +// green in TestVerify_ExpiredCodeIsRetired, whose "row is gone" assertion cannot +// tell a retired row from one this sweep ate. A past cutoff proves the barrier +// just as well, because the password row would be caught by it if it had any +// expiry at all. +func TestPruneOTPExpiredBefore_SparesPasswordWithNullExpiry(t *testing.T) { + ctx := context.Background() + + user := makeUser(ctx, t) + pwd := makePassword(ctx, t, user.ID) + + _, err := store.PruneOTPExpiredBefore(ctx, xtime.UTCNow().Add(-24*time.Hour), 100) + require.NoError(t, err) + + require.True(t, credentialExists(ctx, t, pwd.ID)) + require.Equal(t, 1, countForUser(ctx, t, user.ID)) +} + +// TestPruneOTPExpiredBefore_SparesPasswordWithExpiry is the kind guard, and the +// only test that can prove it exists. Unlike the NULL-expiry case above, this row +// WOULD be deleted by a predicate that dropped `kind = 'otp'`. +func TestPruneOTPExpiredBefore_SparesPasswordWithExpiry(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + user := makeUser(ctx, t) + pwd := makePasswordExpiringAt(ctx, t, user.ID, now.Add(-72*time.Hour)) + + deleted, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + require.GreaterOrEqual(t, deleted, int64(0)) + + require.True(t, credentialExists(ctx, t, pwd.ID), + "a password row must survive even with an expiry older than the cutoff") + require.Equal(t, 1, countForUser(ctx, t, user.ID)) +} + +// TestPruneOTPExpiredBefore_ConsumedRowsFollowTheSameThreshold pins that +// consumed_at plays no part in the predicate: a consumed code leaves on age +// alone, and a consumed-but-recent code stays. +func TestPruneOTPExpiredBefore_ConsumedRowsFollowTheSameThreshold(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + agedUser := makeUser(ctx, t) + recentUser := makeUser(ctx, t) + + aged := makeOTPExpiringAt(ctx, t, agedUser.ID, now.Add(-48*time.Hour)) + recent := makeOTPExpiringAt(ctx, t, recentUser.ID, now.Add(-time.Minute)) + + // consumed_at is excluded from Create's column list by design, so the state + // is reached through ConsumeOTP. It does not check expiry, so an already-aged + // row can be consumed. + for _, id := range []uuid.UUID{aged.ID, recent.ID} { + ok, err := store.ConsumeOTP(ctx, id) + require.NoError(t, err) + require.True(t, ok) + } + + _, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + + require.False(t, credentialExists(ctx, t, aged.ID), + "a consumed code past the cutoff leaves through the same sweep") + require.True(t, credentialExists(ctx, t, recent.ID), + "being consumed grants no early deletion") +} + +// TestPruneOTPExpiredBefore_RespectsLimit pins that one call never removes more +// than the batch bound, which is what makes the service's drain loop meaningful. +func TestPruneOTPExpiredBefore_RespectsLimit(t *testing.T) { + ctx := context.Background() + now := xtime.UTCNow() + + // One OTP per user, because one live OTP per user is a partial-unique + // invariant, so a surviving id is exactly a surviving row. + ids := make([]uuid.UUID, 0, 3) + for range 3 { + u := makeUser(ctx, t) + ids = append(ids, makeOTPExpiringAt(ctx, t, u.ID, now.Add(-48*time.Hour)).ID) + } + + deleted, err := store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 2) + require.NoError(t, err) + require.LessOrEqual(t, deleted, int64(2), "one call must not exceed the limit") + + remaining := 0 + for _, id := range ids { + if credentialExists(ctx, t, id) { + remaining++ + } + } + require.GreaterOrEqual(t, remaining, 1, + "with 3 eligible rows and a limit of 2, at least one must be left for the next batch") + + // Drain the rest so the fixtures do not linger. + _, err = store.PruneOTPExpiredBefore(ctx, now.Add(-24*time.Hour), 100) + require.NoError(t, err) + for _, id := range ids { + require.False(t, credentialExists(ctx, t, id)) + } +} From c02334b375f30e5318bf5f9ba14a7f28bbe55fbe Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 16:39:21 +0300 Subject: [PATCH 3/9] feat(auth): add the retention sweep to the OTP service (RUK-296) Prune drains the store's batch behind a single cutoff, mirroring invitation.Service.Prune, with two deliberate differences. The retention fallback is 24h, not the sibling's 365 days: the direction of safety is inverted here, because a year-long fallback would silently disable the sweep and let code digests and session nonces pile up -- exactly what this job exists to prevent. And the completion line is logged unconditionally, including a zero-row sweep, because this job carries no metric and that line is its only liveness evidence; logging only non-empty sweeps would make a cron that stopped firing indistinguishable from one with nothing to collect. Coercing a non-positive retention is the safety guard rather than hygiene. The per-code attempt ceiling is enforced by the row's continued existence -- claimSlot refuses to issue while it finds a live row with attempts exhausted -- so deleting such a row would hand back a fresh code with a fresh counter and turn "five attempts per code" into "five attempts per code, unlimited codes". Only a non-positive retention can place the cutoff at or after now, so the coercion is what closes it, and it runs before the cutoff is computed. The assertion behind it is unreachable by construction and kept as a guard against a future edit; the fallback must not be narrowed to make it firable. Also adds the generated mock for services/otp, which the Makefile already declared but no commit had produced. Mutation-verified: narrowing the fallback to `retention == 0` fails CutoffIsAlwaysInThePast; removing the batch cap fails StopsAtBatchCap; re-adding the sibling's `if total > 0` fails LogsCompletionEvenWhenNothingDeleted. Co-Authored-By: Claude Opus 5 --- .../generated/mocks/services/otp/service.go | 531 ++++++++++++++++++ internal/services/otp/prune.go | 111 ++++ internal/services/otp/prune_internal_test.go | 197 +++++++ internal/services/otp/service.go | 6 +- 4 files changed, 844 insertions(+), 1 deletion(-) create mode 100644 internal/pkg/generated/mocks/services/otp/service.go create mode 100644 internal/services/otp/prune.go create mode 100644 internal/services/otp/prune_internal_test.go diff --git a/internal/pkg/generated/mocks/services/otp/service.go b/internal/pkg/generated/mocks/services/otp/service.go new file mode 100644 index 00000000..0011504d --- /dev/null +++ b/internal/pkg/generated/mocks/services/otp/service.go @@ -0,0 +1,531 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: ./internal/services/otp/service.go +// +// Generated by this command: +// +// mockgen -typed -destination ./internal/pkg/generated/mocks/services/otp/service.go -source ./internal/services/otp/service.go +// + +// Package mock_otp is a generated GoMock package. +package mock_otp + +import ( + context "context" + reflect "reflect" + time "time" + + uuid "github.com/google/uuid" + entity "github.com/ruko1202/maintmode/internal/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockStore is a mock of Store interface. +type MockStore struct { + ctrl *gomock.Controller + recorder *MockStoreMockRecorder + isgomock struct{} +} + +// MockStoreMockRecorder is the mock recorder for MockStore. +type MockStoreMockRecorder struct { + mock *MockStore +} + +// NewMockStore creates a new mock instance. +func NewMockStore(ctrl *gomock.Controller) *MockStore { + mock := &MockStore{ctrl: ctrl} + mock.recorder = &MockStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockStore) EXPECT() *MockStoreMockRecorder { + return m.recorder +} + +// ClaimOTPAttempt mocks base method. +func (m *MockStore) ClaimOTPAttempt(ctx context.Context, id uuid.UUID, maxAttempts int16) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ClaimOTPAttempt", ctx, id, maxAttempts) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ClaimOTPAttempt indicates an expected call of ClaimOTPAttempt. +func (mr *MockStoreMockRecorder) ClaimOTPAttempt(ctx, id, maxAttempts any) *MockStoreClaimOTPAttemptCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClaimOTPAttempt", reflect.TypeOf((*MockStore)(nil).ClaimOTPAttempt), ctx, id, maxAttempts) + return &MockStoreClaimOTPAttemptCall{Call: call} +} + +// MockStoreClaimOTPAttemptCall wrap *gomock.Call +type MockStoreClaimOTPAttemptCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreClaimOTPAttemptCall) Return(arg0 bool, arg1 error) *MockStoreClaimOTPAttemptCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreClaimOTPAttemptCall) Do(f func(context.Context, uuid.UUID, int16) (bool, error)) *MockStoreClaimOTPAttemptCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreClaimOTPAttemptCall) DoAndReturn(f func(context.Context, uuid.UUID, int16) (bool, error)) *MockStoreClaimOTPAttemptCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// ConsumeOTP mocks base method. +func (m *MockStore) ConsumeOTP(ctx context.Context, id uuid.UUID) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ConsumeOTP", ctx, id) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ConsumeOTP indicates an expected call of ConsumeOTP. +func (mr *MockStoreMockRecorder) ConsumeOTP(ctx, id any) *MockStoreConsumeOTPCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ConsumeOTP", reflect.TypeOf((*MockStore)(nil).ConsumeOTP), ctx, id) + return &MockStoreConsumeOTPCall{Call: call} +} + +// MockStoreConsumeOTPCall wrap *gomock.Call +type MockStoreConsumeOTPCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreConsumeOTPCall) Return(arg0 bool, arg1 error) *MockStoreConsumeOTPCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreConsumeOTPCall) Do(f func(context.Context, uuid.UUID) (bool, error)) *MockStoreConsumeOTPCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreConsumeOTPCall) DoAndReturn(f func(context.Context, uuid.UUID) (bool, error)) *MockStoreConsumeOTPCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// Create mocks base method. +func (m *MockStore) Create(ctx context.Context, cred *entity.AuthCredential) (*entity.AuthCredential, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, cred) + ret0, _ := ret[0].(*entity.AuthCredential) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Create indicates an expected call of Create. +func (mr *MockStoreMockRecorder) Create(ctx, cred any) *MockStoreCreateCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockStore)(nil).Create), ctx, cred) + return &MockStoreCreateCall{Call: call} +} + +// MockStoreCreateCall wrap *gomock.Call +type MockStoreCreateCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreCreateCall) Return(arg0 *entity.AuthCredential, arg1 error) *MockStoreCreateCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreCreateCall) Do(f func(context.Context, *entity.AuthCredential) (*entity.AuthCredential, error)) *MockStoreCreateCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreCreateCall) DoAndReturn(f func(context.Context, *entity.AuthCredential) (*entity.AuthCredential, error)) *MockStoreCreateCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// GetUnconsumedOTPByUserID mocks base method. +func (m *MockStore) GetUnconsumedOTPByUserID(ctx context.Context, userID uuid.UUID) (*entity.AuthCredential, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetUnconsumedOTPByUserID", ctx, userID) + ret0, _ := ret[0].(*entity.AuthCredential) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetUnconsumedOTPByUserID indicates an expected call of GetUnconsumedOTPByUserID. +func (mr *MockStoreMockRecorder) GetUnconsumedOTPByUserID(ctx, userID any) *MockStoreGetUnconsumedOTPByUserIDCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUnconsumedOTPByUserID", reflect.TypeOf((*MockStore)(nil).GetUnconsumedOTPByUserID), ctx, userID) + return &MockStoreGetUnconsumedOTPByUserIDCall{Call: call} +} + +// MockStoreGetUnconsumedOTPByUserIDCall wrap *gomock.Call +type MockStoreGetUnconsumedOTPByUserIDCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreGetUnconsumedOTPByUserIDCall) Return(arg0 *entity.AuthCredential, arg1 error) *MockStoreGetUnconsumedOTPByUserIDCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreGetUnconsumedOTPByUserIDCall) Do(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreGetUnconsumedOTPByUserIDCall) DoAndReturn(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// GetUnconsumedOTPByUserIDForUpdate mocks base method. +func (m *MockStore) GetUnconsumedOTPByUserIDForUpdate(ctx context.Context, userID uuid.UUID) (*entity.AuthCredential, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetUnconsumedOTPByUserIDForUpdate", ctx, userID) + ret0, _ := ret[0].(*entity.AuthCredential) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetUnconsumedOTPByUserIDForUpdate indicates an expected call of GetUnconsumedOTPByUserIDForUpdate. +func (mr *MockStoreMockRecorder) GetUnconsumedOTPByUserIDForUpdate(ctx, userID any) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUnconsumedOTPByUserIDForUpdate", reflect.TypeOf((*MockStore)(nil).GetUnconsumedOTPByUserIDForUpdate), ctx, userID) + return &MockStoreGetUnconsumedOTPByUserIDForUpdateCall{Call: call} +} + +// MockStoreGetUnconsumedOTPByUserIDForUpdateCall wrap *gomock.Call +type MockStoreGetUnconsumedOTPByUserIDForUpdateCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStoreGetUnconsumedOTPByUserIDForUpdateCall) Return(arg0 *entity.AuthCredential, arg1 error) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStoreGetUnconsumedOTPByUserIDForUpdateCall) Do(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStoreGetUnconsumedOTPByUserIDForUpdateCall) DoAndReturn(f func(context.Context, uuid.UUID) (*entity.AuthCredential, error)) *MockStoreGetUnconsumedOTPByUserIDForUpdateCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// PruneOTPExpiredBefore mocks base method. +func (m *MockStore) PruneOTPExpiredBefore(ctx context.Context, cutoff time.Time, limit int64) (int64, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "PruneOTPExpiredBefore", ctx, cutoff, limit) + ret0, _ := ret[0].(int64) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// PruneOTPExpiredBefore indicates an expected call of PruneOTPExpiredBefore. +func (mr *MockStoreMockRecorder) PruneOTPExpiredBefore(ctx, cutoff, limit any) *MockStorePruneOTPExpiredBeforeCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PruneOTPExpiredBefore", reflect.TypeOf((*MockStore)(nil).PruneOTPExpiredBefore), ctx, cutoff, limit) + return &MockStorePruneOTPExpiredBeforeCall{Call: call} +} + +// MockStorePruneOTPExpiredBeforeCall wrap *gomock.Call +type MockStorePruneOTPExpiredBeforeCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockStorePruneOTPExpiredBeforeCall) Return(arg0 int64, arg1 error) *MockStorePruneOTPExpiredBeforeCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockStorePruneOTPExpiredBeforeCall) Do(f func(context.Context, time.Time, int64) (int64, error)) *MockStorePruneOTPExpiredBeforeCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockStorePruneOTPExpiredBeforeCall) DoAndReturn(f func(context.Context, time.Time, int64) (int64, error)) *MockStorePruneOTPExpiredBeforeCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockUserService is a mock of UserService interface. +type MockUserService struct { + ctrl *gomock.Controller + recorder *MockUserServiceMockRecorder + isgomock struct{} +} + +// MockUserServiceMockRecorder is the mock recorder for MockUserService. +type MockUserServiceMockRecorder struct { + mock *MockUserService +} + +// NewMockUserService creates a new mock instance. +func NewMockUserService(ctrl *gomock.Controller) *MockUserService { + mock := &MockUserService{ctrl: ctrl} + mock.recorder = &MockUserServiceMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockUserService) EXPECT() *MockUserServiceMockRecorder { + return m.recorder +} + +// GetByEmail mocks base method. +func (m *MockUserService) GetByEmail(ctx context.Context, email string) (*entity.User, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetByEmail", ctx, email) + ret0, _ := ret[0].(*entity.User) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetByEmail indicates an expected call of GetByEmail. +func (mr *MockUserServiceMockRecorder) GetByEmail(ctx, email any) *MockUserServiceGetByEmailCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetByEmail", reflect.TypeOf((*MockUserService)(nil).GetByEmail), ctx, email) + return &MockUserServiceGetByEmailCall{Call: call} +} + +// MockUserServiceGetByEmailCall wrap *gomock.Call +type MockUserServiceGetByEmailCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockUserServiceGetByEmailCall) Return(arg0 *entity.User, arg1 error) *MockUserServiceGetByEmailCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockUserServiceGetByEmailCall) Do(f func(context.Context, string) (*entity.User, error)) *MockUserServiceGetByEmailCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockUserServiceGetByEmailCall) DoAndReturn(f func(context.Context, string) (*entity.User, error)) *MockUserServiceGetByEmailCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockKeyring is a mock of Keyring interface. +type MockKeyring struct { + ctrl *gomock.Controller + recorder *MockKeyringMockRecorder + isgomock struct{} +} + +// MockKeyringMockRecorder is the mock recorder for MockKeyring. +type MockKeyringMockRecorder struct { + mock *MockKeyring +} + +// NewMockKeyring creates a new mock instance. +func NewMockKeyring(ctrl *gomock.Controller) *MockKeyring { + mock := &MockKeyring{ctrl: ctrl} + mock.recorder = &MockKeyringMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockKeyring) EXPECT() *MockKeyringMockRecorder { + return m.recorder +} + +// WrapDEK mocks base method. +func (m *MockKeyring) WrapDEK(dek []byte) ([]byte, string, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WrapDEK", dek) + ret0, _ := ret[0].([]byte) + ret1, _ := ret[1].(string) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// WrapDEK indicates an expected call of WrapDEK. +func (mr *MockKeyringMockRecorder) WrapDEK(dek any) *MockKeyringWrapDEKCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WrapDEK", reflect.TypeOf((*MockKeyring)(nil).WrapDEK), dek) + return &MockKeyringWrapDEKCall{Call: call} +} + +// MockKeyringWrapDEKCall wrap *gomock.Call +type MockKeyringWrapDEKCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockKeyringWrapDEKCall) Return(wrapped []byte, kekID string, err error) *MockKeyringWrapDEKCall { + c.Call = c.Call.Return(wrapped, kekID, err) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockKeyringWrapDEKCall) Do(f func([]byte) ([]byte, string, error)) *MockKeyringWrapDEKCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockKeyringWrapDEKCall) DoAndReturn(f func([]byte) ([]byte, string, error)) *MockKeyringWrapDEKCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockCipher is a mock of Cipher interface. +type MockCipher struct { + ctrl *gomock.Controller + recorder *MockCipherMockRecorder + isgomock struct{} +} + +// MockCipherMockRecorder is the mock recorder for MockCipher. +type MockCipherMockRecorder struct { + mock *MockCipher +} + +// NewMockCipher creates a new mock instance. +func NewMockCipher(ctrl *gomock.Controller) *MockCipher { + mock := &MockCipher{ctrl: ctrl} + mock.recorder = &MockCipherMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockCipher) EXPECT() *MockCipherMockRecorder { + return m.recorder +} + +// Encrypt mocks base method. +func (m *MockCipher) Encrypt(dek, plaintext, aad []byte) ([]byte, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Encrypt", dek, plaintext, aad) + ret0, _ := ret[0].([]byte) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Encrypt indicates an expected call of Encrypt. +func (mr *MockCipherMockRecorder) Encrypt(dek, plaintext, aad any) *MockCipherEncryptCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Encrypt", reflect.TypeOf((*MockCipher)(nil).Encrypt), dek, plaintext, aad) + return &MockCipherEncryptCall{Call: call} +} + +// MockCipherEncryptCall wrap *gomock.Call +type MockCipherEncryptCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockCipherEncryptCall) Return(arg0 []byte, arg1 error) *MockCipherEncryptCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockCipherEncryptCall) Do(f func([]byte, []byte, []byte) ([]byte, error)) *MockCipherEncryptCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockCipherEncryptCall) DoAndReturn(f func([]byte, []byte, []byte) ([]byte, error)) *MockCipherEncryptCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + +// MockTaskScheduler is a mock of TaskScheduler interface. +type MockTaskScheduler struct { + ctrl *gomock.Controller + recorder *MockTaskSchedulerMockRecorder + isgomock struct{} +} + +// MockTaskSchedulerMockRecorder is the mock recorder for MockTaskScheduler. +type MockTaskSchedulerMockRecorder struct { + mock *MockTaskScheduler +} + +// NewMockTaskScheduler creates a new mock instance. +func NewMockTaskScheduler(ctrl *gomock.Controller) *MockTaskScheduler { + mock := &MockTaskScheduler{ctrl: ctrl} + mock.recorder = &MockTaskSchedulerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockTaskScheduler) EXPECT() *MockTaskSchedulerMockRecorder { + return m.recorder +} + +// Schedule mocks base method. +func (m *MockTaskScheduler) Schedule(ctx context.Context, taskType string, payload any, idempotencyKey string) (uuid.UUID, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Schedule", ctx, taskType, payload, idempotencyKey) + ret0, _ := ret[0].(uuid.UUID) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Schedule indicates an expected call of Schedule. +func (mr *MockTaskSchedulerMockRecorder) Schedule(ctx, taskType, payload, idempotencyKey any) *MockTaskSchedulerScheduleCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Schedule", reflect.TypeOf((*MockTaskScheduler)(nil).Schedule), ctx, taskType, payload, idempotencyKey) + return &MockTaskSchedulerScheduleCall{Call: call} +} + +// MockTaskSchedulerScheduleCall wrap *gomock.Call +type MockTaskSchedulerScheduleCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockTaskSchedulerScheduleCall) Return(arg0 uuid.UUID, arg1 error) *MockTaskSchedulerScheduleCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockTaskSchedulerScheduleCall) Do(f func(context.Context, string, any, string) (uuid.UUID, error)) *MockTaskSchedulerScheduleCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockTaskSchedulerScheduleCall) DoAndReturn(f func(context.Context, string, any, string) (uuid.UUID, error)) *MockTaskSchedulerScheduleCall { + c.Call = c.Call.DoAndReturn(f) + return c +} diff --git a/internal/services/otp/prune.go b/internal/services/otp/prune.go new file mode 100644 index 00000000..911dd82a --- /dev/null +++ b/internal/services/otp/prune.go @@ -0,0 +1,111 @@ +package otp + +import ( + "context" + "time" + + "github.com/ruko1202/xlog" + "github.com/ruko1202/xlog/xfield" + + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +// maxPruneBatches caps how many batches one Prune call runs, so a single sweep +// can never loop unbounded. The periodic job runs again on the next tick, so any +// leftover eligible rows are drained then; this only bounds per-invocation work. +const maxPruneBatches = 100 + +// defaultPruneBatchLimit is the per-statement DELETE bound used when the caller +// passes a non-positive batchLimit, so the drain loop's "deleted < batchLimit" +// stop condition stays meaningful. +const defaultPruneBatchLimit = 1000 + +// defaultPruneRetention is the age threshold used when the caller passes a +// non-positive retention. +// +// It is 24h, NOT the 365 days its invitation counterpart uses, and the direction +// of safety is why. A one-time code lives minutes, so a day is already a wide +// margin for looking into a failed sign-in; a year-long fallback would silently +// disable the sweep and let code digests and session nonces pile up -- the exact +// thing this job exists to prevent. It must stay equal to the value shipped in +// deployment config, so that an unset retention behaves like the configured one. +const defaultPruneRetention = 24 * time.Hour + +// Prune deletes one-time codes whose expires_at is older than the retention +// window, in bounded batches. Consumed and merely-expired codes leave through +// this same sweep -- consumed_at is not part of the predicate -- and password +// credentials are never eligible. +// +// batchLimit bounds one DELETE so the per-statement lock footprint stays small; +// Prune loops batches until one comes back short (the table is drained for this +// cutoff) or the per-call batch cap is hit. Both tunables come from the cron task +// payload (derived from config). +// +// The cutoff is computed once at call time so every batch in this sweep targets +// the same instant -- rows aging past the threshold mid-sweep wait for the next +// tick rather than shifting the boundary under the loop. +// +// SAFETY. Coercing a non-positive retention is not cosmetic. The per-code attempt +// ceiling is enforced by the row's continued existence: claimSlot refuses to +// issue a new code while it finds a live row with the attempts exhausted, so +// deleting such a row would hand back a fresh code with a fresh counter and turn +// "five attempts per code" into "five attempts per code, unlimited codes". A +// non-positive retention is the only way to reach a cutoff at or after now, and +// hence the only way to make a live row eligible -- so the coercion below is what +// closes that hole. Every positive retention is safe however small, because a row +// is eligible only once its expires_at is already past. +func (s *Service) Prune(ctx context.Context, retention time.Duration, batchLimit int64) error { + ctx, span := xlog.WithOperationSpan(ctx, "service.OTP.Prune") + defer span.End() + + if batchLimit <= 0 { + batchLimit = defaultPruneBatchLimit + } + if retention <= 0 { + retention = defaultPruneRetention + } + + cutoff := xtime.UTCNow().Add(-retention) + + // Unreachable given the coercion above, and kept deliberately: it is a guard + // against a future edit narrowing that fallback or reordering these steps, + // not against any input reachable today. Do not "fix" the fallback to make + // this branch testable -- refusing a zero retention instead of defaulting it + // would break the documented "unset means default" contract. + if !cutoff.Before(xtime.UTCNow()) { + xlog.Error(ctx, "refusing to prune one-time codes with a cutoff that is not in the past", + xfield.Time("cutoff", cutoff), + xfield.Duration("retention", retention), + ) + return nil + } + + var total int64 + for range maxPruneBatches { + deleted, err := s.store.PruneOTPExpiredBefore(ctx, cutoff, batchLimit) + if err != nil { + xlog.Error(ctx, "failed to prune expired one-time codes batch", + xfield.Time("cutoff", cutoff), + xfield.Int64("prunedSoFar", total), + xfield.Error(err), + ) + return err + } + + total += deleted + if deleted < batchLimit { + break + } + } + + // Logged unconditionally, including a zero-row sweep -- a knowing divergence + // from the invitation sweep, which logs only when it deleted something. This + // job has no metric, so the line is the only evidence it ran: without it a + // cron that stopped firing looks exactly like a cron with nothing to do. + xlog.Info(ctx, "pruned expired one-time codes", + xfield.Int64("count", total), + xfield.Time("cutoff", cutoff), + ) + + return nil +} diff --git a/internal/services/otp/prune_internal_test.go b/internal/services/otp/prune_internal_test.go new file mode 100644 index 00000000..6219b84f --- /dev/null +++ b/internal/services/otp/prune_internal_test.go @@ -0,0 +1,197 @@ +package otp + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ruko1202/xlog" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + mock_otp "github.com/ruko1202/maintmode/internal/pkg/generated/mocks/services/otp" + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +// The drain loop is exercised over a mocked store: the batch sizes a real +// database would return are exactly what these tests need to script, and the +// predicate itself is covered by the store's own integration tests. +func newPruneService(t *testing.T) (*Service, *mock_otp.MockStore) { + t.Helper() + + store := mock_otp.NewMockStore(gomock.NewController(t)) + + return &Service{store: store}, store +} + +func TestPrune_DrainsUntilShortBatch(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + gomock.InOrder( + store.EXPECT().PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(2)).Return(int64(2), nil), + store.EXPECT().PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(2)).Return(int64(2), nil), + // A batch shorter than the limit means the table is drained for this + // cutoff, so the loop must stop here rather than call again. + store.EXPECT().PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(2)).Return(int64(1), nil), + ) + + require.NoError(t, svc.Prune(context.Background(), 24*time.Hour, 2)) +} + +// TestPrune_StopsAtBatchCap pins that one invocation cannot loop unbounded: with +// every batch coming back full, the loop still ends after maxPruneBatches. +func TestPrune_StopsAtBatchCap(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), int64(10)). + Return(int64(10), nil). + Times(maxPruneBatches) + + require.NoError(t, svc.Prune(context.Background(), 24*time.Hour, 10)) +} + +// TestPrune_UsesOneCutoffForEveryBatch pins that the boundary is computed once, +// so rows aging mid-sweep wait for the next tick instead of shifting the cutoff +// under the loop. +func TestPrune_UsesOneCutoffForEveryBatch(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + var seen []time.Time + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, cutoff time.Time, limit int64) (int64, error) { + seen = append(seen, cutoff) + if len(seen) < 3 { + return limit, nil + } + return 0, nil + }). + Times(3) + + require.NoError(t, svc.Prune(context.Background(), 24*time.Hour, 5)) + require.Len(t, seen, 3) + require.Equal(t, seen[0], seen[1]) + require.Equal(t, seen[0], seen[2]) +} + +// TestPrune_CutoffIsAlwaysInThePast is the safety property: whatever retention +// reaches Prune -- including the non-positive values the fallback absorbs -- the +// store must never be handed a cutoff at or after now. A future cutoff would make +// a live, attempt-exhausted code eligible and reset the guess ceiling that +// claimSlot enforces by the row's continued existence. +func TestPrune_CutoffIsAlwaysInThePast(t *testing.T) { + t.Parallel() + + for _, retention := range []time.Duration{ + -365 * 24 * time.Hour, + -time.Second, + 0, + time.Minute, + 24 * time.Hour, + } { + t.Run(retention.String(), func(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + var got time.Time + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, cutoff time.Time, _ int64) (int64, error) { + got = cutoff + return 0, nil + }) + + require.NoError(t, svc.Prune(context.Background(), retention, 10)) + require.True(t, got.Before(xtime.UTCNow()), + "cutoff %s must be strictly in the past", got) + }) + } +} + +// TestPrune_NonPositiveTunablesFallBack pins both fallbacks. The retention one +// matters most: a non-positive value must become the 24h default rather than a +// cutoff at or past now. +func TestPrune_NonPositiveTunablesFallBack(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + var gotCutoff time.Time + var gotLimit int64 + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, cutoff time.Time, limit int64) (int64, error) { + gotCutoff, gotLimit = cutoff, limit + return 0, nil + }) + + before := xtime.UTCNow() + require.NoError(t, svc.Prune(context.Background(), 0, 0)) + + require.Equal(t, int64(defaultPruneBatchLimit), gotLimit) + // The cutoff must sit a default-retention behind now, give or take the + // microseconds the call itself takes. + require.WithinDuration(t, before.Add(-defaultPruneRetention), gotCutoff, time.Minute) +} + +func TestPrune_PropagatesStoreError(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + + wantErr := errors.New("delete failed") + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + Return(int64(0), wantErr) + + require.ErrorIs(t, svc.Prune(context.Background(), 24*time.Hour, 10), wantErr) +} + +// observedPruneCtx returns a context whose logger writes into the returned sink, +// mirroring the seam bootstrapauth's tests use. +func observedPruneCtx(t *testing.T) (context.Context, *observer.ObservedLogs) { + t.Helper() + + core, logs := observer.New(zapcore.DebugLevel) + + return xlog.ContextWithLogger(context.Background(), xlog.NewZapAdapter(zap.New(core))), logs +} + +// TestPrune_LogsCompletionEvenWhenNothingDeleted defends the one place this +// sweep knowingly diverges from its invitation sibling, which wraps the same +// line in `if total > 0`. +// +// The job carries no metric, so this line is the only evidence it ran. Restoring +// the sibling's condition in the name of consistency would make a cron that +// stopped firing indistinguishable from a cron with nothing to collect -- and +// with nothing asserting on it, that regression would be silent. +func TestPrune_LogsCompletionEvenWhenNothingDeleted(t *testing.T) { + t.Parallel() + + svc, store := newPruneService(t) + store.EXPECT(). + PruneOTPExpiredBefore(gomock.Any(), gomock.Any(), gomock.Any()). + Return(int64(0), nil) + + ctx, logs := observedPruneCtx(t) + require.NoError(t, svc.Prune(ctx, 24*time.Hour, 10)) + + done := logs.FilterMessage("pruned expired one-time codes").All() + require.Len(t, done, 1, "a zero-row sweep must still report that it ran") + require.Equal(t, zapcore.InfoLevel, done[0].Level) + + require.Equal(t, int64(0), done[0].ContextMap()["count"], + "the count field is what makes the line useful") +} diff --git a/internal/services/otp/service.go b/internal/services/otp/service.go index d41f440b..7521d226 100644 --- a/internal/services/otp/service.go +++ b/internal/services/otp/service.go @@ -20,7 +20,7 @@ import ( ) // Store is the credential persistence this service needs. Defined consumer-side -// so the service depends on the three operations it performs, not on the whole +// so the service depends only on the operations it performs, not on the whole // store, and so tests can substitute a fake. type Store interface { Create(ctx context.Context, cred *entity.AuthCredential) (*entity.AuthCredential, error) @@ -32,6 +32,10 @@ type Store interface { // is no lock to hold and nothing for it to protect against. GetUnconsumedOTPByUserID(ctx context.Context, userID uuid.UUID) (*entity.AuthCredential, error) ClaimOTPAttempt(ctx context.Context, id uuid.UUID, maxAttempts int16) (bool, error) + // PruneOTPExpiredBefore backs the retention sweep -- the one operation here + // that is not part of a sign-in, since it deletes spent codes rather than + // reading or retiring one. + PruneOTPExpiredBefore(ctx context.Context, cutoff time.Time, limit int64) (int64, error) } // UserService resolves the address to a user. Only the lookup is needed. From 53f9132ab0cacde3907f59b175e279c464749150 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 16:39:37 +0300 Subject: [PATCH 4/9] feat(auth): schedule the otp.prune sweep (RUK-296) The processor package (day-bucketed external id, typed processor, cron factory), its registration in the goque worker, and the otp_prune block in all four stands. The cron spec falls back to "15 3 * * *" instead of failing to build the job. The sibling sweeps abort startup when their spec is missing, which in this merged auth+maintmode process means an absent config line takes sign-in down; falling back the way license.heartbeat already does keeps that to a degraded schedule, while a malformed spec still surfaces as an error. The 03:15 offset also keeps this sweep out of the 03:00 minute the three other daily jobs share. Retention is 24h in config, matching the service fallback and the factory's cmp.Or default -- three values that must agree, and the factory test asserts the literal rather than only its local constant so they cannot drift apart silently. The YAML comment deliberately does not repeat the siblings' "must be set or NewTaskProcessors panics" warning: with the fallback in place that sentence is false for this block. It documents the ~48h worst-case residency instead, since retention alone reads as a tighter guarantee than the daily cadence gives. registerOTPPrune is extracted for symmetry with registerInvitationRotation, not for funlen headroom. The registration test covers what the entity drift test structurally cannot: adding otp.prune to both task-type lists while never registering the periodic job satisfies every assertion in that file. Mutation-verified -- dropping the RegisterPeriodicJob call or the cmp.Or fallback each fail a test. Verified by booting the binary against local config with no otp_prune block: both goque-processor-otp.prune and goque-periodic-job-otp.prune.cron start and the startup coverage guard passes. Co-Authored-By: Claude Opus 5 --- Makefile | 1 + deployment/maintmode/dev/app.config.yaml | 16 +++ deployment/maintmode/local/app.config.yaml | 16 +++ deployment/maintmode/prod/app.config.yaml | 16 +++ deployment/maintmode/test/app.config.yaml | 16 +++ .../bootstrap/otp_prune_registration_test.go | 54 ++++++++ internal/app/bootstrap/processors.go | 53 ++++++- .../otppruneprocessor/external_id.go | 21 +++ .../otppruneprocessor/processor.go | 71 ++++++++++ .../otppruneprocessor/processor_test.go | 130 ++++++++++++++++++ .../otppruneprocessor/processor.go | 80 +++++++++++ 11 files changed, 472 insertions(+), 2 deletions(-) create mode 100644 internal/app/bootstrap/otp_prune_registration_test.go create mode 100644 internal/goque_processors/otppruneprocessor/external_id.go create mode 100644 internal/goque_processors/otppruneprocessor/processor.go create mode 100644 internal/goque_processors/otppruneprocessor/processor_test.go create mode 100644 internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go diff --git a/Makefile b/Makefile index 63a0be87..7c547cf8 100644 --- a/Makefile +++ b/Makefile @@ -280,6 +280,7 @@ mocks: $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/invitationrotateprocessor/processor.go -source ./internal/goque_processors/invitationrotateprocessor/processor.go $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/invitationpruneprocessor/processor.go -source ./internal/goque_processors/invitationpruneprocessor/processor.go $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/licenseheartbeatprocessor/processor.go -source ./internal/goque_processors/licenseheartbeatprocessor/processor.go + $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go -source ./internal/goque_processors/otppruneprocessor/processor.go $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/services/usersummary/service.go -source ./internal/services/usersummary/service.go $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/services/license/service.go -source ./internal/services/license/service.go $(GOBIN)/mockgen -typed -destination ./internal/pkg/generated/mocks/server/middlewares/auth.go -source ./internal/server/middlewares/auth.go diff --git a/deployment/maintmode/dev/app.config.yaml b/deployment/maintmode/dev/app.config.yaml index 70336cfe..4f1c1f03 100644 --- a/deployment/maintmode/dev/app.config.yaml +++ b/deployment/maintmode/dev/app.config.yaml @@ -193,6 +193,22 @@ task_processor: # 365 days by created_at. Terminal rows older than this are deleted in batches. retention: 8760h batch_limit: 1000 + # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at + # is older than retention. Consumed and merely-expired codes leave through the + # same threshold; password rows are never touched. + # + # Retention is short (unlike the year the other sweeps keep) because the row + # holds a code digest and a session nonce while the code itself lives minutes. + # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. + # + # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and + # invitation_prune an empty value degrades the schedule instead of aborting + # startup. The 03:15 offset keeps this sweep out of the minute the three other + # daily jobs share. + otp_prune: + cron_spec: "15 3 * * *" + retention: 24h + batch_limit: 1000 # crypto addresses the master key (KEK) that wraps the data-encryption keys # protecting integration secrets at rest. A KEK is addressed by URI: diff --git a/deployment/maintmode/local/app.config.yaml b/deployment/maintmode/local/app.config.yaml index b3a358eb..6913d811 100644 --- a/deployment/maintmode/local/app.config.yaml +++ b/deployment/maintmode/local/app.config.yaml @@ -185,6 +185,22 @@ task_processor: cron_spec: "0 3 * * *" retention: 8760h batch_limit: 1000 + # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at + # is older than retention. Consumed and merely-expired codes leave through the + # same threshold; password rows are never touched. + # + # Retention is short (unlike the year the other sweeps keep) because the row + # holds a code digest and a session nonce while the code itself lives minutes. + # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. + # + # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and + # invitation_prune an empty value degrades the schedule instead of aborting + # startup. The 03:15 offset keeps this sweep out of the minute the three other + # daily jobs share. + otp_prune: + cron_spec: "15 3 * * *" + retention: 24h + batch_limit: 1000 # crypto addresses the master key (KEK) that wraps the data-encryption keys # protecting integration secrets at rest. A KEK is addressed by URI: diff --git a/deployment/maintmode/prod/app.config.yaml b/deployment/maintmode/prod/app.config.yaml index c7e2ec8d..62f7aa8e 100644 --- a/deployment/maintmode/prod/app.config.yaml +++ b/deployment/maintmode/prod/app.config.yaml @@ -227,6 +227,22 @@ task_processor: cron_spec: "0 3 * * *" retention: 8760h batch_limit: 1000 + # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at + # is older than retention. Consumed and merely-expired codes leave through the + # same threshold; password rows are never touched. + # + # Retention is short (unlike the year the other sweeps keep) because the row + # holds a code digest and a session nonce while the code itself lives minutes. + # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. + # + # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and + # invitation_prune an empty value degrades the schedule instead of aborting + # startup. The 03:15 offset keeps this sweep out of the minute the three other + # daily jobs share. + otp_prune: + cron_spec: "15 3 * * *" + retention: 24h + batch_limit: 1000 # crypto addresses the master key (KEK) that wraps the data-encryption keys # protecting integration secrets at rest. A KEK is addressed by URI: diff --git a/deployment/maintmode/test/app.config.yaml b/deployment/maintmode/test/app.config.yaml index 1307bd93..f49beb68 100644 --- a/deployment/maintmode/test/app.config.yaml +++ b/deployment/maintmode/test/app.config.yaml @@ -193,6 +193,22 @@ task_processor: cron_spec: "0 3 * * *" retention: 8760h batch_limit: 1000 + # otp_prune deletes spent one-time codes: rows with kind='otp' whose expires_at + # is older than retention. Consumed and merely-expired codes leave through the + # same threshold; password rows are never touched. + # + # Retention is short (unlike the year the other sweeps keep) because the row + # holds a code digest and a session nonce while the code itself lives minutes. + # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. + # + # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and + # invitation_prune an empty value degrades the schedule instead of aborting + # startup. The 03:15 offset keeps this sweep out of the minute the three other + # daily jobs share. + otp_prune: + cron_spec: "15 3 * * *" + retention: 24h + batch_limit: 1000 # crypto addresses the master key (KEK) that wraps the data-encryption keys # protecting integration secrets at rest. A KEK is addressed by URI: diff --git a/internal/app/bootstrap/otp_prune_registration_test.go b/internal/app/bootstrap/otp_prune_registration_test.go new file mode 100644 index 00000000..bf2f4918 --- /dev/null +++ b/internal/app/bootstrap/otp_prune_registration_test.go @@ -0,0 +1,54 @@ +package bootstrap + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/ruko1202/maintmode/internal/config" + "github.com/ruko1202/maintmode/internal/entity" +) + +// registerOTPPrune is exercised directly rather than through NewTaskProcessors, +// which needs live stores and services. What matters here is the pair it +// registers: the entity drift test cannot see a missing .cron type -- adding +// otp.prune to both task-type lists while never registering its periodic job +// satisfies every assertion in that file -- so this is the only place the cron +// half is actually pinned. +func TestRegisterOTPPrune_RegistersBothHalves(t *testing.T) { + reg := newProcessorRegistrar(newRegisterOnlyGoque()) + + // A nil Services is enough: registration stores the processor, it does not + // call the pruner. + require.NoError(t, registerOTPPrune(reg, config.TaskProcessorConfig{ + OTPPrune: config.TaskProcessorOTPPruneConfig{CronSpec: "15 3 * * *"}, + }, &Services{})) + + require.Contains(t, reg.registered, entity.ProcessorTaskOTPPrune, + "the task processor must be registered or enqueued sweeps linger undrained") + require.Contains(t, reg.registered, entity.ProcessorTaskOTPPruneCron, + "the periodic job must be registered or the sweep never fires") +} + +// TestRegisterOTPPrune_EmptyCronSpecStillBoots pins the deliberate divergence +// from the sibling sweeps: they abort startup when their spec is missing, which +// in this merged process would take sign-in down over an absent config line. +func TestRegisterOTPPrune_EmptyCronSpecStillBoots(t *testing.T) { + reg := newProcessorRegistrar(newRegisterOnlyGoque()) + + require.NoError(t, registerOTPPrune(reg, config.TaskProcessorConfig{}, &Services{})) + require.Contains(t, reg.registered, entity.ProcessorTaskOTPPruneCron) +} + +// TestRegisterOTPPrune_RejectsInvalidCronSpec is the other half: a value that is +// present but malformed is an operator error and must surface, not be silently +// replaced by the default. +func TestRegisterOTPPrune_RejectsInvalidCronSpec(t *testing.T) { + reg := newProcessorRegistrar(newRegisterOnlyGoque()) + + err := registerOTPPrune(reg, config.TaskProcessorConfig{ + OTPPrune: config.TaskProcessorOTPPruneConfig{CronSpec: "not a cron spec"}, + }, &Services{}) + require.Error(t, err) + require.Contains(t, err.Error(), "otp-prune") +} diff --git a/internal/app/bootstrap/processors.go b/internal/app/bootstrap/processors.go index c78469bf..411f18b7 100644 --- a/internal/app/bootstrap/processors.go +++ b/internal/app/bootstrap/processors.go @@ -15,6 +15,7 @@ import ( "github.com/ruko1202/maintmode/internal/goque_processors/invitationrotateprocessor" "github.com/ruko1202/maintmode/internal/goque_processors/licenseheartbeatprocessor" "github.com/ruko1202/maintmode/internal/goque_processors/otpemailprocessor" + "github.com/ruko1202/maintmode/internal/goque_processors/otppruneprocessor" "github.com/ruko1202/maintmode/internal/goque_processors/reminderprocessor" "github.com/ruko1202/maintmode/internal/pkg/secrets" "github.com/ruko1202/maintmode/internal/services/otp" @@ -23,10 +24,17 @@ import ( "github.com/ruko1202/maintmode/internal/entity" ) +// defaultOTPPruneCronSpec is the schedule used when config omits one. It is +// offset from the 03:00 the three other daily sweeps share so four drain loops +// do not start in the same minute. +const defaultOTPPruneCronSpec = "15 3 * * *" + // NewTaskProcessors builds the single goque worker for the maintmode process and // registers every task type the merged process owns: maint.reminder, -// maint.auto.cancel (+ cron), invitation.email, audit.write and audit.prune -// (+ cron), plus license.heartbeat (+ cron) when SaaS license mode is enabled. +// maint.auto.cancel (+ cron), invitation.email, otp.email, audit.write, +// audit.prune (+ cron), the invitation rotate/prune pair (+ crons) and +// otp.prune (+ cron), plus license.heartbeat (+ cron) when SaaS license mode is +// enabled. // Everything is registered on one registrar, and verify() runs once at the end to // assert the registered set matches entity.ExpectedProcessorTaskTypes for this // process's toggles. @@ -184,6 +192,10 @@ func NewTaskProcessors( reg.RegisterPeriodicJob(heartbeatJob) } + if err := registerOTPPrune(reg, cfg, services); err != nil { + return nil, err + } + if err := reg.verify(entity.ExpectedProcessorTaskTypes(licenseEnabled)); err != nil { return nil, err } @@ -191,6 +203,43 @@ func NewTaskProcessors( return goq, nil } +// registerOTPPrune registers the one-time-code retention sweep: a daily cron job +// enqueues one task carrying the retention window and batch limit from config, +// and the processor deletes auth_credentials rows with kind='otp' whose +// expires_at is older than that window, in bounded batches. Consumed codes age +// out through the same predicate; password rows are never eligible. +// +// One worker, because the sweep is a single drained DELETE loop that must not run +// concurrently with itself, and a day-bucketed external id, so multi-replica +// ticks collapse to one enqueue per day — the same shape as the other sweeps. +// +// Unlike them, the cron spec has a code-side default. audit.prune and +// invitation.prune abort startup when their spec is missing, which for a merged +// process means sign-in goes down over an absent config line; falling back the +// way license.heartbeat does keeps a missing value to a degraded schedule. +func registerOTPPrune(reg *processorRegistrar, cfg config.TaskProcessorConfig, services *Services) error { + pruneCfg := cfg.OTPPrune + + reg.RegisterProcessor( + entity.ProcessorTaskOTPPrune, + otppruneprocessor.NewTaskProcessor(services.OTP), + messagingProcessorOpts(cfg.Messaging, 1)..., + ) + + pruneJob, err := goque.NewCronJob( + entity.ProcessorTaskOTPPruneCron, + cmp.Or(pruneCfg.CronSpec, defaultOTPPruneCronSpec), + time.UTC, + otppruneprocessor.NewTaskFactory(pruneCfg.Retention, pruneCfg.BatchLimit), + ) + if err != nil { + return fmt.Errorf("failed to build otp-prune cron job: %w", err) + } + reg.RegisterPeriodicJob(pruneJob) + + return nil +} + // registerInvitationRotation registers the invitation lifecycle's daily // retention pair on reg: // diff --git a/internal/goque_processors/otppruneprocessor/external_id.go b/internal/goque_processors/otppruneprocessor/external_id.go new file mode 100644 index 00000000..37050e54 --- /dev/null +++ b/internal/goque_processors/otppruneprocessor/external_id.go @@ -0,0 +1,21 @@ +package otppruneprocessor + +import ( + "fmt" + "time" +) + +// otpPruneExternalID derives the deterministic, day-bucketed external id for an +// otp.prune task. Truncating to the day makes every replica that ticks within +// the same day produce the same id, so the goque (type, external_id) unique +// constraint dedupes them to a single enqueued task per day. Retention is a +// daily-granularity concern, so one prune per day is the intended cadence even +// if the cron schedule fires more often. +// +// The bucket is computed from wall-clock fire time, so replicas with clock skew +// straddling a day boundary can land in adjacent buckets and enqueue twice. That +// is harmless: the prune is idempotent (a second run on the same cutoff finds +// nothing left to delete) and bounded. +func otpPruneExternalID(now time.Time) string { + return fmt.Sprintf("otp-prune-%s", now.UTC().Format("2006-01-02")) +} diff --git a/internal/goque_processors/otppruneprocessor/processor.go b/internal/goque_processors/otppruneprocessor/processor.go new file mode 100644 index 00000000..8c2e6837 --- /dev/null +++ b/internal/goque_processors/otppruneprocessor/processor.go @@ -0,0 +1,71 @@ +// Package otppruneprocessor handles otp.prune goque tasks. The task is produced +// once per cron tick by a periodic job and carries the retention tunables +// (window + batch limit) in its payload; at process time it deletes one-time +// codes whose expires_at is older than the retention window, in bounded batches. +package otppruneprocessor + +import ( + "cmp" + "context" + "time" + + "github.com/ruko1202/goque" + "github.com/ruko1202/xlog" + + "github.com/ruko1202/maintmode/internal/entity" + "github.com/ruko1202/maintmode/internal/utils/xtime" +) + +const ( + // Both defaults must match the ones the service falls back to and the values + // shipped in deployment config, so an unset payload field behaves the same + // wherever it is resolved. 24h is short on purpose: the row holds a code + // digest and a session nonce, and the code itself lives only minutes. + defaultRetention = 24 * time.Hour + defaultBatchLimit = 1000 +) + +// Pruner deletes one-time codes older than the retention window. Defined +// consumer-side so the processor can be tested with a mock. +type Pruner interface { + Prune(ctx context.Context, retention time.Duration, batchLimit int64) error +} + +// NewTaskProcessor returns the goque TaskProcessor for ProcessorTaskOTPPrune. It +// reads the retention window and batch limit from the task payload and runs the +// prune. +func NewTaskProcessor(pruner Pruner) goque.TaskProcessor { + return goque.NewTypedTaskProcessor( + goque.TypedTaskProcessorFunc[entity.ProcessorTaskPayloadOTPPrune]( + func(ctx context.Context, task *goque.TypedTask[entity.ProcessorTaskPayloadOTPPrune]) error { + ctx, span := xlog.WithOperationSpan(ctx, "service.OTP.PruneProcessor.ProcessTask") + defer span.End() + + return pruner.Prune(ctx, task.Payload.Retention, task.Payload.BatchLimit) + }, + ), + goque.WithCancelTaskWhenPayloadDecodeError[entity.ProcessorTaskPayloadOTPPrune](), + ) +} + +// NewTaskFactory returns the goque PeriodicJobFactory that produces one otp.prune +// task per cron tick, stamping the configured retention window and batch limit +// into the payload. +// +// The external id is bucketed to the day so that, with several replicas all +// ticking the same schedule, only the first insert for a given day succeeds; the +// others collide on the (type, external_id) unique key and return +// goque.ErrDuplicateTask -- the desired at-most-once-per-day fan-in. On a single +// replica this never fires. +func NewTaskFactory(retention time.Duration, batchLimit int64) goque.PeriodicJobFactory { + return func(_ context.Context) (*goque.Task, error) { + return goque.NewTaskWithPayloadAndExternalID( + entity.ProcessorTaskOTPPrune, + entity.ProcessorTaskPayloadOTPPrune{ + Retention: cmp.Or(retention, defaultRetention), + BatchLimit: cmp.Or(batchLimit, defaultBatchLimit), + }, + otpPruneExternalID(xtime.UTCNow()), + ) + } +} diff --git a/internal/goque_processors/otppruneprocessor/processor_test.go b/internal/goque_processors/otppruneprocessor/processor_test.go new file mode 100644 index 00000000..66599040 --- /dev/null +++ b/internal/goque_processors/otppruneprocessor/processor_test.go @@ -0,0 +1,130 @@ +package otppruneprocessor + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ruko1202/goque" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/ruko1202/maintmode/internal/entity" + mock_otppruneprocessor "github.com/ruko1202/maintmode/internal/pkg/generated/mocks/goque_processors/otppruneprocessor" +) + +func newPruneTask(t *testing.T, retention time.Duration, batchLimit int64) *goque.Task { + t.Helper() + + task, err := goque.NewTaskWithPayloadAndExternalID( + entity.ProcessorTaskOTPPrune, + entity.ProcessorTaskPayloadOTPPrune{Retention: retention, BatchLimit: batchLimit}, + "test-external-id", + ) + require.NoError(t, err) + + return task +} + +func TestProcessTask_DelegatesWithPayloadTunables(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + var gotRetention time.Duration + var gotLimit int64 + pruner.EXPECT(). + Prune(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, retention time.Duration, batchLimit int64) error { + gotRetention, gotLimit = retention, batchLimit + return nil + }) + + err := NewTaskProcessor(pruner).ProcessTask(context.Background(), newPruneTask(t, 24*time.Hour, 1000)) + require.NoError(t, err) + require.Equal(t, 24*time.Hour, gotRetention) + require.Equal(t, int64(1000), gotLimit) +} + +func TestProcessTask_PropagatesError(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + wantErr := errors.New("prune failed") + pruner.EXPECT().Prune(gomock.Any(), gomock.Any(), gomock.Any()).Return(wantErr) + + err := NewTaskProcessor(pruner).ProcessTask(context.Background(), newPruneTask(t, time.Hour, 10)) + require.ErrorIs(t, err, wantErr) +} + +// TestNewTaskFactory_StampsTunablesAndDayBucket asserts the factory produces a +// well-typed task whose payload survives the JSON round-trip and whose external +// id is day-bucketed. +func TestNewTaskFactory_StampsTunablesAndDayBucket(t *testing.T) { + t.Parallel() + + task, err := NewTaskFactory(48*time.Hour, 500)(context.Background()) + require.NoError(t, err) + require.NotNil(t, task) + require.Equal(t, entity.ProcessorTaskOTPPrune, task.Type) + require.NotEmpty(t, task.ExternalID) + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + var gotRetention time.Duration + var gotLimit int64 + pruner.EXPECT(). + Prune(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, retention time.Duration, batchLimit int64) error { + gotRetention, gotLimit = retention, batchLimit + return nil + }) + + require.NoError(t, NewTaskProcessor(pruner).ProcessTask(context.Background(), task)) + require.Equal(t, 48*time.Hour, gotRetention) + require.Equal(t, int64(500), gotLimit) +} + +// TestNewTaskFactory_DefaultsZeroTunables checks the cmp.Or fallbacks, and pins +// that the retention default is the 24h the service and the deployment config +// also use -- three values that must not drift apart. +func TestNewTaskFactory_DefaultsZeroTunables(t *testing.T) { + t.Parallel() + + task, err := NewTaskFactory(0, 0)(context.Background()) + require.NoError(t, err) + + ctrl := gomock.NewController(t) + pruner := mock_otppruneprocessor.NewMockPruner(ctrl) + + var gotRetention time.Duration + var gotLimit int64 + pruner.EXPECT(). + Prune(gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, retention time.Duration, batchLimit int64) error { + gotRetention, gotLimit = retention, batchLimit + return nil + }) + + require.NoError(t, NewTaskProcessor(pruner).ProcessTask(context.Background(), task)) + require.Equal(t, defaultRetention, gotRetention) + require.Equal(t, 24*time.Hour, gotRetention) + require.Equal(t, int64(defaultBatchLimit), gotLimit) +} + +func TestOTPPruneExternalID_DayBucketed(t *testing.T) { + t.Parallel() + + base := time.Date(2026, 9, 5, 3, 15, 0, 0, time.UTC) + require.Equal(t, "otp-prune-2026-09-05", otpPruneExternalID(base)) + + // Any other moment in the same UTC day must collapse to the same id -- that + // is what dedupes replicas ticking the same schedule. + require.Equal(t, otpPruneExternalID(base), otpPruneExternalID(base.Add(20*time.Hour))) + require.NotEqual(t, otpPruneExternalID(base), otpPruneExternalID(base.Add(24*time.Hour))) +} diff --git a/internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go b/internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go new file mode 100644 index 00000000..f41f79f8 --- /dev/null +++ b/internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go @@ -0,0 +1,80 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: ./internal/goque_processors/otppruneprocessor/processor.go +// +// Generated by this command: +// +// mockgen -typed -destination ./internal/pkg/generated/mocks/goque_processors/otppruneprocessor/processor.go -source ./internal/goque_processors/otppruneprocessor/processor.go +// + +// Package mock_otppruneprocessor is a generated GoMock package. +package mock_otppruneprocessor + +import ( + context "context" + reflect "reflect" + time "time" + + gomock "go.uber.org/mock/gomock" +) + +// MockPruner is a mock of Pruner interface. +type MockPruner struct { + ctrl *gomock.Controller + recorder *MockPrunerMockRecorder + isgomock struct{} +} + +// MockPrunerMockRecorder is the mock recorder for MockPruner. +type MockPrunerMockRecorder struct { + mock *MockPruner +} + +// NewMockPruner creates a new mock instance. +func NewMockPruner(ctrl *gomock.Controller) *MockPruner { + mock := &MockPruner{ctrl: ctrl} + mock.recorder = &MockPrunerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockPruner) EXPECT() *MockPrunerMockRecorder { + return m.recorder +} + +// Prune mocks base method. +func (m *MockPruner) Prune(ctx context.Context, retention time.Duration, batchLimit int64) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Prune", ctx, retention, batchLimit) + ret0, _ := ret[0].(error) + return ret0 +} + +// Prune indicates an expected call of Prune. +func (mr *MockPrunerMockRecorder) Prune(ctx, retention, batchLimit any) *MockPrunerPruneCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Prune", reflect.TypeOf((*MockPruner)(nil).Prune), ctx, retention, batchLimit) + return &MockPrunerPruneCall{Call: call} +} + +// MockPrunerPruneCall wrap *gomock.Call +type MockPrunerPruneCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockPrunerPruneCall) Return(arg0 error) *MockPrunerPruneCall { + c.Call = c.Call.Return(arg0) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockPrunerPruneCall) Do(f func(context.Context, time.Duration, int64) error) *MockPrunerPruneCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockPrunerPruneCall) DoAndReturn(f func(context.Context, time.Duration, int64) error) *MockPrunerPruneCall { + c.Call = c.Call.DoAndReturn(f) + return c +} From f94ffea8122681d9ae2ce6aa3a1cf4e7d2f8cee3 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 21:33:05 +0300 Subject: [PATCH 5/9] fix(auth): drop the bogus cutoff assertion from the otp sweep (RUK-296) The guard read the clock a second time, so it compared the cutoff against a later instant than the one the cutoff was derived from. What it actually tested was "retention exceeds the gap between two clock reads" -- tens of microseconds -- rather than "the cutoff is in the past". That makes it useless for the single regression it was added to catch. If the retention fallback were narrowed to `retention == 0`, a zero retention would produce a cutoff a few microseconds behind now, the branch would evaluate false, and the sweep would proceed exactly as if the guard were absent. Its `return nil` also reported success while doing nothing. A guard that looks like protection but is not is worse than no guard, because the next reader trusts it instead of the coercion that does the work. The regression stays covered by TestPrune_CutoffIsAlwaysInThePast, which asserts on the cutoff the store is actually handed across negative, zero and positive retentions -- verified still dead by re-running that mutation after this change. Co-Authored-By: Claude Opus 5 --- internal/services/otp/prune.go | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/internal/services/otp/prune.go b/internal/services/otp/prune.go index 911dd82a..328874d7 100644 --- a/internal/services/otp/prune.go +++ b/internal/services/otp/prune.go @@ -67,19 +67,6 @@ func (s *Service) Prune(ctx context.Context, retention time.Duration, batchLimit cutoff := xtime.UTCNow().Add(-retention) - // Unreachable given the coercion above, and kept deliberately: it is a guard - // against a future edit narrowing that fallback or reordering these steps, - // not against any input reachable today. Do not "fix" the fallback to make - // this branch testable -- refusing a zero retention instead of defaulting it - // would break the documented "unset means default" contract. - if !cutoff.Before(xtime.UTCNow()) { - xlog.Error(ctx, "refusing to prune one-time codes with a cutoff that is not in the past", - xfield.Time("cutoff", cutoff), - xfield.Duration("retention", retention), - ) - return nil - } - var total int64 for range maxPruneBatches { deleted, err := s.store.PruneOTPExpiredBefore(ctx, cutoff, batchLimit) From b740a2a47e5c05174b1bf47c94d2d259f97d9377 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 21:36:47 +0300 Subject: [PATCH 6/9] test(auth): drive the real constructor and drop echo asserts (RUK-296) The registration tests called registerOTPPrune directly, which left the wiring itself untested: removing the registerOTPPrune call from NewTaskProcessors kept all three green while the sweep would silently never run in the binary. Nothing else covered it either -- the entity drift test only reads the task-type maps, and no test called NewTaskProcessors at all. Verified by mutation: with the call removed the new test fails on "missing=[otp.prune otp.prune.cron]" while the helper-level test still passes. Zero-valued Stores and Services are enough to drive it, so the real constructor needs no database. It also exercises the startup coverage guard, which until now was only ever checked by booting the binary by hand. In the processor test, drop `require.NotNil(task)` after a nil error and `require.NotEmpty(task.ExternalID)`: both are goque echoing its own arguments, and the external id's real value is pinned by TestOTPPruneExternalID_DayBucketed. The task-type assert is KEPT and now carries a comment saying why -- mutating the factory to stamp the .cron type is otherwise undetected, and such a task would sit in the queue with no processor to drain it. Co-Authored-By: Claude Opus 5 --- .../bootstrap/otp_prune_registration_test.go | 46 +++++++++++++++---- .../otppruneprocessor/processor_test.go | 16 +++++-- 2 files changed, 49 insertions(+), 13 deletions(-) diff --git a/internal/app/bootstrap/otp_prune_registration_test.go b/internal/app/bootstrap/otp_prune_registration_test.go index bf2f4918..72be9365 100644 --- a/internal/app/bootstrap/otp_prune_registration_test.go +++ b/internal/app/bootstrap/otp_prune_registration_test.go @@ -9,17 +9,47 @@ import ( "github.com/ruko1202/maintmode/internal/entity" ) -// registerOTPPrune is exercised directly rather than through NewTaskProcessors, -// which needs live stores and services. What matters here is the pair it -// registers: the entity drift test cannot see a missing .cron type -- adding -// otp.prune to both task-type lists while never registering its periodic job -// satisfies every assertion in that file -- so this is the only place the cron -// half is actually pinned. +// baseTaskProcessorConfig is the minimum config NewTaskProcessors needs to build +// every cron job it owns. The sibling sweeps have no code-side spec fallback, so +// theirs must be set or the constructor fails before it reaches otp.prune. +func baseTaskProcessorConfig() config.TaskProcessorConfig { + return config.TaskProcessorConfig{ + MaintAutoCancel: config.TaskProcessorMaintAutoCancelConfig{CronSpec: "* * * * *"}, + AuditPrune: config.TaskProcessorAuditPruneConfig{CronSpec: "0 3 * * *"}, + InvitationRotate: config.TaskProcessorInvitationRotateConfig{CronSpec: "0 3 * * *"}, + InvitationPrune: config.TaskProcessorInvitationPruneConfig{CronSpec: "0 3 * * *"}, + } +} + +// TestNewTaskProcessors_RegistersOTPPrune drives the REAL constructor, not the +// registerOTPPrune helper directly. +// +// That distinction is the whole point of this test. Calling the helper would +// prove only that the helper works, leaving the wiring untested: drop the +// registerOTPPrune call from NewTaskProcessors and a helper-level test stays +// green while the sweep silently never runs. Nothing else would catch it either +// -- the entity drift test only reads the task-type maps, and no other test +// calls NewTaskProcessors at all. +// +// Zero-valued Stores and Services are enough: registration stores the processor +// and never calls into it, and goque accepts a nil TaskStorage for a registrar +// that is not draining. +func TestNewTaskProcessors_RegistersOTPPrune(t *testing.T) { + cfg := baseTaskProcessorConfig() + cfg.OTPPrune = config.TaskProcessorOTPPruneConfig{CronSpec: "15 3 * * *"} + + goq, err := NewTaskProcessors(cfg, config.LicenseConfig{}, config.Auth{}, &Stores{}, &Services{}) + require.NoError(t, err) + require.NotNil(t, goq) +} + +// TestRegisterOTPPrune_RegistersBothHalves pins which types the helper registers. +// The entity drift test cannot see a missing .cron entry -- adding otp.prune to +// both task-type lists while never registering its periodic job satisfies every +// assertion in that file -- so this is the only place the cron half is named. func TestRegisterOTPPrune_RegistersBothHalves(t *testing.T) { reg := newProcessorRegistrar(newRegisterOnlyGoque()) - // A nil Services is enough: registration stores the processor, it does not - // call the pruner. require.NoError(t, registerOTPPrune(reg, config.TaskProcessorConfig{ OTPPrune: config.TaskProcessorOTPPruneConfig{CronSpec: "15 3 * * *"}, }, &Services{})) diff --git a/internal/goque_processors/otppruneprocessor/processor_test.go b/internal/goque_processors/otppruneprocessor/processor_test.go index 66599040..3a133a0d 100644 --- a/internal/goque_processors/otppruneprocessor/processor_test.go +++ b/internal/goque_processors/otppruneprocessor/processor_test.go @@ -61,17 +61,23 @@ func TestProcessTask_PropagatesError(t *testing.T) { require.ErrorIs(t, err, wantErr) } -// TestNewTaskFactory_StampsTunablesAndDayBucket asserts the factory produces a -// well-typed task whose payload survives the JSON round-trip and whose external -// id is day-bucketed. +// TestNewTaskFactory_StampsTunablesAndDayBucket asserts the tunables the factory +// stamps survive the JSON round-trip and reach the pruner unchanged. +// +// The task type is asserted because the factory chooses which constant to stamp, +// and stamping the wrong one -- the .cron producer type, say -- would enqueue +// tasks that no registered processor drains, losing the work silently. That is +// verified by mutation, not assumed. +// +// Nothing else about the task struct is asserted: NotNil after a nil error, or a +// non-empty external id, is goque echoing its own arguments back. The external +// id's actual value is pinned by TestOTPPruneExternalID_DayBucketed. func TestNewTaskFactory_StampsTunablesAndDayBucket(t *testing.T) { t.Parallel() task, err := NewTaskFactory(48*time.Hour, 500)(context.Background()) require.NoError(t, err) - require.NotNil(t, task) require.Equal(t, entity.ProcessorTaskOTPPrune, task.Type) - require.NotEmpty(t, task.ExternalID) ctrl := gomock.NewController(t) pruner := mock_otppruneprocessor.NewMockPruner(ctrl) From 42087fdd96cbcb6b09d278e08efa007ee0b738e3 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 21:44:26 +0300 Subject: [PATCH 7/9] fix(auth): take the otp.prune cron spec from config only (RUK-296) Drop the cmp.Or fallback and the defaultOTPPruneCronSpec constant: the spec now comes straight from config, and an empty or malformed value fails NewTaskProcessors and aborts startup, exactly as it does for audit.prune and invitation.prune. The fallback was defending against a stand whose config predated this change, but that trade was wrong: a value that is set but malformed should fail loudly rather than be silently replaced, and a sweep running on a schedule nobody configured is harder to notice than a process that refuses to start. All four stands set the value, so the failure mode is reachable only by deleting a line that is already there. Also removes otp_prune_registration_test.go. Its remaining tests either called registerOTPPrune and then asserted that registerOTPPrune had registered something -- a tautology -- or duplicated what the constructor-level checks already cover. The YAML comments no longer promise a fallback that does not exist; they now carry the same "must be set" warning as the neighbouring blocks. Co-Authored-By: Claude Opus 5 --- deployment/maintmode/dev/app.config.yaml | 7 +- deployment/maintmode/local/app.config.yaml | 7 +- deployment/maintmode/prod/app.config.yaml | 7 +- deployment/maintmode/test/app.config.yaml | 7 +- .../bootstrap/otp_prune_registration_test.go | 84 ------------------- internal/app/bootstrap/processors.go | 14 +--- 6 files changed, 16 insertions(+), 110 deletions(-) delete mode 100644 internal/app/bootstrap/otp_prune_registration_test.go diff --git a/deployment/maintmode/dev/app.config.yaml b/deployment/maintmode/dev/app.config.yaml index 4f1c1f03..7e34cb51 100644 --- a/deployment/maintmode/dev/app.config.yaml +++ b/deployment/maintmode/dev/app.config.yaml @@ -201,10 +201,9 @@ task_processor: # holds a code digest and a session nonce while the code itself lives minutes. # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. # - # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and - # invitation_prune an empty value degrades the schedule instead of aborting - # startup. The 03:15 offset keeps this sweep out of the minute the three other - # daily jobs share. + # cron_spec must be set here: like audit_prune and invitation_prune, an empty + # or malformed value fails NewTaskProcessors and aborts startup. The 03:15 + # offset keeps this sweep out of the minute the three other daily jobs share. otp_prune: cron_spec: "15 3 * * *" retention: 24h diff --git a/deployment/maintmode/local/app.config.yaml b/deployment/maintmode/local/app.config.yaml index 6913d811..dccf9dcf 100644 --- a/deployment/maintmode/local/app.config.yaml +++ b/deployment/maintmode/local/app.config.yaml @@ -193,10 +193,9 @@ task_processor: # holds a code digest and a session nonce while the code itself lives minutes. # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. # - # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and - # invitation_prune an empty value degrades the schedule instead of aborting - # startup. The 03:15 offset keeps this sweep out of the minute the three other - # daily jobs share. + # cron_spec must be set here: like audit_prune and invitation_prune, an empty + # or malformed value fails NewTaskProcessors and aborts startup. The 03:15 + # offset keeps this sweep out of the minute the three other daily jobs share. otp_prune: cron_spec: "15 3 * * *" retention: 24h diff --git a/deployment/maintmode/prod/app.config.yaml b/deployment/maintmode/prod/app.config.yaml index 62f7aa8e..c7e2ca27 100644 --- a/deployment/maintmode/prod/app.config.yaml +++ b/deployment/maintmode/prod/app.config.yaml @@ -235,10 +235,9 @@ task_processor: # holds a code digest and a session nonce while the code itself lives minutes. # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. # - # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and - # invitation_prune an empty value degrades the schedule instead of aborting - # startup. The 03:15 offset keeps this sweep out of the minute the three other - # daily jobs share. + # cron_spec must be set here: like audit_prune and invitation_prune, an empty + # or malformed value fails NewTaskProcessors and aborts startup. The 03:15 + # offset keeps this sweep out of the minute the three other daily jobs share. otp_prune: cron_spec: "15 3 * * *" retention: 24h diff --git a/deployment/maintmode/test/app.config.yaml b/deployment/maintmode/test/app.config.yaml index f49beb68..e5d97bba 100644 --- a/deployment/maintmode/test/app.config.yaml +++ b/deployment/maintmode/test/app.config.yaml @@ -201,10 +201,9 @@ task_processor: # holds a code digest and a session nonce while the code itself lives minutes. # Worst-case residency is retention plus one cron period, so ~48h here, not 24h. # - # cron_spec has a code-side default ("15 3 * * *"), so unlike audit_prune and - # invitation_prune an empty value degrades the schedule instead of aborting - # startup. The 03:15 offset keeps this sweep out of the minute the three other - # daily jobs share. + # cron_spec must be set here: like audit_prune and invitation_prune, an empty + # or malformed value fails NewTaskProcessors and aborts startup. The 03:15 + # offset keeps this sweep out of the minute the three other daily jobs share. otp_prune: cron_spec: "15 3 * * *" retention: 24h diff --git a/internal/app/bootstrap/otp_prune_registration_test.go b/internal/app/bootstrap/otp_prune_registration_test.go deleted file mode 100644 index 72be9365..00000000 --- a/internal/app/bootstrap/otp_prune_registration_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package bootstrap - -import ( - "testing" - - "github.com/stretchr/testify/require" - - "github.com/ruko1202/maintmode/internal/config" - "github.com/ruko1202/maintmode/internal/entity" -) - -// baseTaskProcessorConfig is the minimum config NewTaskProcessors needs to build -// every cron job it owns. The sibling sweeps have no code-side spec fallback, so -// theirs must be set or the constructor fails before it reaches otp.prune. -func baseTaskProcessorConfig() config.TaskProcessorConfig { - return config.TaskProcessorConfig{ - MaintAutoCancel: config.TaskProcessorMaintAutoCancelConfig{CronSpec: "* * * * *"}, - AuditPrune: config.TaskProcessorAuditPruneConfig{CronSpec: "0 3 * * *"}, - InvitationRotate: config.TaskProcessorInvitationRotateConfig{CronSpec: "0 3 * * *"}, - InvitationPrune: config.TaskProcessorInvitationPruneConfig{CronSpec: "0 3 * * *"}, - } -} - -// TestNewTaskProcessors_RegistersOTPPrune drives the REAL constructor, not the -// registerOTPPrune helper directly. -// -// That distinction is the whole point of this test. Calling the helper would -// prove only that the helper works, leaving the wiring untested: drop the -// registerOTPPrune call from NewTaskProcessors and a helper-level test stays -// green while the sweep silently never runs. Nothing else would catch it either -// -- the entity drift test only reads the task-type maps, and no other test -// calls NewTaskProcessors at all. -// -// Zero-valued Stores and Services are enough: registration stores the processor -// and never calls into it, and goque accepts a nil TaskStorage for a registrar -// that is not draining. -func TestNewTaskProcessors_RegistersOTPPrune(t *testing.T) { - cfg := baseTaskProcessorConfig() - cfg.OTPPrune = config.TaskProcessorOTPPruneConfig{CronSpec: "15 3 * * *"} - - goq, err := NewTaskProcessors(cfg, config.LicenseConfig{}, config.Auth{}, &Stores{}, &Services{}) - require.NoError(t, err) - require.NotNil(t, goq) -} - -// TestRegisterOTPPrune_RegistersBothHalves pins which types the helper registers. -// The entity drift test cannot see a missing .cron entry -- adding otp.prune to -// both task-type lists while never registering its periodic job satisfies every -// assertion in that file -- so this is the only place the cron half is named. -func TestRegisterOTPPrune_RegistersBothHalves(t *testing.T) { - reg := newProcessorRegistrar(newRegisterOnlyGoque()) - - require.NoError(t, registerOTPPrune(reg, config.TaskProcessorConfig{ - OTPPrune: config.TaskProcessorOTPPruneConfig{CronSpec: "15 3 * * *"}, - }, &Services{})) - - require.Contains(t, reg.registered, entity.ProcessorTaskOTPPrune, - "the task processor must be registered or enqueued sweeps linger undrained") - require.Contains(t, reg.registered, entity.ProcessorTaskOTPPruneCron, - "the periodic job must be registered or the sweep never fires") -} - -// TestRegisterOTPPrune_EmptyCronSpecStillBoots pins the deliberate divergence -// from the sibling sweeps: they abort startup when their spec is missing, which -// in this merged process would take sign-in down over an absent config line. -func TestRegisterOTPPrune_EmptyCronSpecStillBoots(t *testing.T) { - reg := newProcessorRegistrar(newRegisterOnlyGoque()) - - require.NoError(t, registerOTPPrune(reg, config.TaskProcessorConfig{}, &Services{})) - require.Contains(t, reg.registered, entity.ProcessorTaskOTPPruneCron) -} - -// TestRegisterOTPPrune_RejectsInvalidCronSpec is the other half: a value that is -// present but malformed is an operator error and must surface, not be silently -// replaced by the default. -func TestRegisterOTPPrune_RejectsInvalidCronSpec(t *testing.T) { - reg := newProcessorRegistrar(newRegisterOnlyGoque()) - - err := registerOTPPrune(reg, config.TaskProcessorConfig{ - OTPPrune: config.TaskProcessorOTPPruneConfig{CronSpec: "not a cron spec"}, - }, &Services{}) - require.Error(t, err) - require.Contains(t, err.Error(), "otp-prune") -} diff --git a/internal/app/bootstrap/processors.go b/internal/app/bootstrap/processors.go index 411f18b7..d56fb317 100644 --- a/internal/app/bootstrap/processors.go +++ b/internal/app/bootstrap/processors.go @@ -24,11 +24,6 @@ import ( "github.com/ruko1202/maintmode/internal/entity" ) -// defaultOTPPruneCronSpec is the schedule used when config omits one. It is -// offset from the 03:00 the three other daily sweeps share so four drain loops -// do not start in the same minute. -const defaultOTPPruneCronSpec = "15 3 * * *" - // NewTaskProcessors builds the single goque worker for the maintmode process and // registers every task type the merged process owns: maint.reminder, // maint.auto.cancel (+ cron), invitation.email, otp.email, audit.write, @@ -213,10 +208,9 @@ func NewTaskProcessors( // concurrently with itself, and a day-bucketed external id, so multi-replica // ticks collapse to one enqueue per day — the same shape as the other sweeps. // -// Unlike them, the cron spec has a code-side default. audit.prune and -// invitation.prune abort startup when their spec is missing, which for a merged -// process means sign-in goes down over an absent config line; falling back the -// way license.heartbeat does keeps a missing value to a degraded schedule. +// The cron spec comes straight from config with no code-side default: a missing +// or malformed value fails here and aborts startup, exactly as it does for +// audit.prune and invitation.prune. All four deployment stands set it. func registerOTPPrune(reg *processorRegistrar, cfg config.TaskProcessorConfig, services *Services) error { pruneCfg := cfg.OTPPrune @@ -228,7 +222,7 @@ func registerOTPPrune(reg *processorRegistrar, cfg config.TaskProcessorConfig, s pruneJob, err := goque.NewCronJob( entity.ProcessorTaskOTPPruneCron, - cmp.Or(pruneCfg.CronSpec, defaultOTPPruneCronSpec), + pruneCfg.CronSpec, time.UTC, otppruneprocessor.NewTaskFactory(pruneCfg.Retention, pruneCfg.BatchLimit), ) From 1ca7f252fed2ecd238352a0e444d29652b3fba01 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 21:47:58 +0300 Subject: [PATCH 8/9] test(auth): drop the otp retention validator test (RUK-296) validateOTPRetention is a four-line negative check whose sibling, validateInvitationRetention, ships without a test of its own. Matching that is the more consistent choice. Nothing safety-relevant rides on it: the behaviour that matters -- a non-positive retention never reaching the DELETE -- is pinned at the service level by TestPrune_CutoffIsAlwaysInThePast, where it is actually reachable. The startup validator is a config-hygiene message layered on top of that. Co-Authored-By: Claude Opus 5 --- internal/config/app_config_otp_prune_test.go | 65 -------------------- 1 file changed, 65 deletions(-) delete mode 100644 internal/config/app_config_otp_prune_test.go diff --git a/internal/config/app_config_otp_prune_test.go b/internal/config/app_config_otp_prune_test.go deleted file mode 100644 index a58d1aae..00000000 --- a/internal/config/app_config_otp_prune_test.go +++ /dev/null @@ -1,65 +0,0 @@ -package config - -import ( - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -// The validator is exercised directly rather than through LoadAppConfig: the -// initConfig chain reports a validation failure with log.Panicf, which would -// take the test binary down instead of failing an assertion. -func TestValidateOTPRetention(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - retention time.Duration - wantErr bool - }{ - { - name: "negative is rejected", - retention: -time.Hour, - wantErr: true, - }, - { - // Zero means "unset": the service applies its own default, so it must - // not be treated as a misconfiguration. - name: "zero is accepted as unset", - retention: 0, - wantErr: false, - }, - { - name: "shipped default is accepted", - retention: 24 * time.Hour, - wantErr: false, - }, - { - // A retention far below the code lifetime is a poor forensic window - // but not a safety problem: a row is only eligible once expires_at is - // already in the past, so no live code can be reached. Accepting it - // keeps the validator a mirror of its invitation sibling. - name: "small positive is accepted", - retention: time.Minute, - wantErr: false, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - - cfg := &AppConfig{} - cfg.TaskProcessor.OTPPrune.Retention = tt.retention - - err := cfg.validateOTPRetention() - if tt.wantErr { - require.Error(t, err) - require.Contains(t, err.Error(), "task_processor.otp_prune.retention") - return - } - require.NoError(t, err) - }) - } -} From 3abbead968ea419ebcf2e41ce075469d30e03846 Mon Sep 17 00:00:00 2001 From: Ruslan Kosykh Date: Sat, 5 Sep 2026 21:57:08 +0300 Subject: [PATCH 9/9] docs(auth): remove the otp sign-in runbook (RUK-296) Nothing referenced it, neither of the two shipped periodic sweeps has an equivalent, and the only mention of runbooks anywhere in the repo is a comment in alerts.yml phrased as "if you keep operator runbooks of your own" -- which reads as an invitation, not as an established practice. It also carried facts that duplicate the config and the code comments (the schedule, the retention window, which rows the predicate spares), so it was set up to drift out of date the first time either changed. Dropping it here rather than extending it with an otp.prune section, which is what an earlier commit on this branch did before the file's place in the repo was questioned. Co-Authored-By: Claude Opus 5 --- docs/runbooks/otp-signin.md | 108 ------------------------------------ 1 file changed, 108 deletions(-) delete mode 100644 docs/runbooks/otp-signin.md diff --git a/docs/runbooks/otp-signin.md b/docs/runbooks/otp-signin.md deleted file mode 100644 index c049c3be..00000000 --- a/docs/runbooks/otp-signin.md +++ /dev/null @@ -1,108 +0,0 @@ -# Runbook: email one-time-code sign-in - -Covers `POST /api/v1/login/otp/request` and `POST /api/v1/login/otp/verify`. - -Both endpoints answer every failure identically on purpose, so a user's report -("it says the code is wrong") rarely narrows anything down. The audit trail and -the logs are where the distinctions live. - -## Limiter degradation - -Three tiers guard these routes, and they degrade differently when Valkey is -unreachable: - -| Tier | Key | On a Valkey outage | -| --- | --- | --- | -| Per IP | client address | per-replica bucket, so the effective cap is `N × limit` across `N` replicas | -| Per address | the email in the body | same, per replica | -| Instance-wide | a constant | **stops applying entirely** | - -The instance-wide tier is the deliberate exception. Its key is a constant, so a -per-replica bucket would put every caller into one in-memory bucket and let a -single attacker hold the whole sign-in surface at 429. Losing the anti-sweep -control during an outage beats converting the outage into a total sign-in -outage; the per-address tier, which protects an individual account, keeps -working throughout. - -The practical consequence: **during a Valkey outage this deployment is more -exposed to a distributed sweep than usual.** If an outage coincides with a spike -in failed verifications, treat the two as related. - -## Reading the fallback signal - -The alert is `RateLimiterValkeyFallback`, on `ratelimit_valkey_failures_total`. - -That counter carries **no tier label**, so it cannot tell you which limiter fell -back. Since one Valkey outage is the common cause for all three, read a firing -alert as *all three degraded at once* — per-IP and per-address to `N × limit`, -instance-wide to nothing. - -A sustained non-zero rate means Valkey, not an attack. Check Valkey's own health -first; the limiters recover on their own once it answers. - -`otp_attempt_claim_errors_total` is separate and more serious. It counts -failures to claim a guess against a code, and the claim *is* the per-code -attempt ceiling. A sustained non-zero rate means the brute-force ceiling is not -being enforced — verification fails closed while this happens, so users see -errors rather than a silent hole, but the cause is a database problem and -nothing else will say so. - -## Clearing the burnt-code barrier - -Spending every guess on a code does **not** free the active-code slot: no new -code is issued until the burnt one expires (at most `auth.otp_ttl`, 5 minutes by -default). Without that rule, "five attempts" would mean "five attempts per code, -unlimited codes". - -The refusal is deliberately invisible — the same `202` and the same response -shape as a successful request — and deliberately unaudited, since no secret was -presented and auditing it would let anyone write unbounded rows by replaying the -endpoint. So a barred user has no self-service diagnosis, and this is the only -way to confirm it. - -Find a barred address: - -```sql -SELECT c.id, u.email, c.attempts, c.expires_at -FROM auth_credentials c -JOIN users u ON u.id = c.user_id -WHERE c.kind = 'otp' - AND c.consumed_at IS NULL - AND c.expires_at > now() - AND c.attempts >= 5 -- auth.otp_max_attempts - AND u.email = 'user@example.com'; -``` - -Clear it, **keyed on the id from that query**: - -```sql -UPDATE auth_credentials SET consumed_at = now() WHERE id = ''; -``` - -The predicate is not optional. Running the `SET` without a `WHERE` retires every -live code on the instance: nobody is signed in by it, and everyone waiting on a -code has to request a new one. - -Usually the right answer is to wait. The barrier is at most one `otp_ttl`, and -clearing it by hand for a user who is being targeted removes the control while -the attempt is in progress. - -## Reading a 429 storm - -A burst of 429s on `/login/otp` no longer means "one caller is hammering us". -With an instance-wide tier, one attacker exhausting the global budget returns -429 to **every** caller, including legitimate ones. Nothing in the metrics -distinguishes that from a local burst or from a Valkey fallback. - -To tell them apart: - -- Are the 429s spread across many client addresses? That points at the global - tier, so a sweep is in progress and everyone is being refused. -- Concentrated on one address? Per-IP or per-address, and the ordinary cause. -- Is `RateLimiterValkeyFallback` firing? Then caps are per-replica and the - instance-wide tier is off — the 429s are not coming from it. - -Any alert added on this route later must exclude 429. The per-IP limiter carries -no route component, so `login/oauth`, `login/password` and `users/invitations` -share one budget with these routes and produce 429s here for traffic that never -touched them.