diff --git a/docs/configuration.md b/docs/configuration.md index 55537dbc06..e95301413c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -319,13 +319,30 @@ token: # "memory" – in-process mutex (default, single-replica only) # "postgres" – PostgreSQL lease-table (multi-replica) backend: memory + # memory section is read only when backend == "memory". + memory: + # acquireDeadline bounds how long one acquisition waits for an EID held + # by another anchor. Every backend bounds its own waiting so that a + # caller which passes no deadline still gets an answer, and so that + # spending the whole budget can be reported as such; auditor.lock does + # not retry an acquisition that already exhausted it. Defaults to the + # same 1m as the postgres backend, so switching backends does not + # silently change how long an audit can block. + acquireDeadline: 1m # postgres section is read only when backend == "postgres". postgres: # ttl is the lease duration for each EID lock row. ttl: 30s - # acquireBackoff is the wait between retry attempts when a lock is contended. + # acquireBackoff is the initial wait between retry attempts when a lock + # is contended. Successive waits grow exponentially and are jittered, + # so this is the floor rather than a fixed poll interval. acquireBackoff: 100ms - # acquireDeadline is the total time allowed to acquire all EID locks. + # acquireMaxBackoff caps that growth. Raised to acquireBackoff if set + # below it. + acquireMaxBackoff: 2s + # acquireDeadline is the total time allowed to acquire all EID locks, + # and the whole budget for waiting out contention: auditor.lock does + # not retry an acquisition that already exhausted it. acquireDeadline: 1m # heartbeat is the interval at which held leases are renewed (~TTL/3). heartbeat: 10s @@ -810,6 +827,14 @@ Default values: - **backoffMultiplier**: Factor by which the backoff delay increases after each retry (exponential growth) - **jitterFactor**: Randomization factor (0.0 to 1.0) added to backoff delays to prevent multiple auditors from retrying simultaneously (prevents thundering herd problem) +**Relationship to `auditor.locker`:** this retry covers failures that another attempt +might survive, such as a transient database error. It deliberately does **not** retry +an acquisition that failed with `ErrLockAcquireTimeout`, because that means the locker +already spent its own `acquireDeadline` waiting out the contention — retrying would +spend it again. Waiting under contention is configured once, in +[`auditor.locker`](#optional-tokentmsauditorlocker), so `maxRetries` here does not +multiply `acquireDeadline` there. + **Tuning Recommendations:** 1. **For High-Contention Environments:** @@ -924,6 +949,7 @@ token: postgres: ttl: 30s acquireBackoff: 100ms + acquireMaxBackoff: 2s acquireDeadline: 1m heartbeat: 10s owner: @@ -933,8 +959,9 @@ Default values: - backend: `memory` (in-process mutex, single-replica only) - postgres.ttl: 30s -- postgres.acquireBackoff: 100ms -- postgres.acquireDeadline: 1m +- postgres.acquireBackoff: 100ms (initial wait; grows exponentially, jittered) +- postgres.acquireMaxBackoff: 2s (cap on that growth; raised to `acquireBackoff` if set below it) +- postgres.acquireDeadline: 1m (the whole budget for waiting out contention) - postgres.heartbeat: 10s - postgres.owner: empty, defaults to the FSC node ID (`config.Provider.ID()`). Required when `backend: postgres` — if both this value and `fsc.id` are empty or blank, the diff --git a/docs/services/auditor.md b/docs/services/auditor.md index 1dd3d9f828..ba8939f5b2 100644 --- a/docs/services/auditor.md +++ b/docs/services/auditor.md @@ -83,6 +83,37 @@ When multiple auditor replicas share the same AuditDB (PostgreSQL), concurrent p The locker is injected into `auditdb.StoreService` at startup. Before appending audit records, the store calls `AssertLocksHeld` to verify leases are still valid. +Both backends implement the same contract, defined on the `Locker` interface, because the backend is chosen from configuration — a behavioural difference between them would be a correctness difference between two deployments running the same code. The shared expectations are exercised against every backend by `locker/conformance_test.go`. + +**`AssertLocksHeld` detects lost locks, not absent ones.** It fails only when a lease this replica held has since expired or been taken over. An anchor that holds nothing succeeds — whether because the request yielded no enrollment IDs, or because the caller never locked anything for it. The latter is a supported flow: an auditor may call `Validate` (which takes no locks) and then `Append`, as the dvp and nft sample views do. + +**`AcquireLocks` is all-or-nothing, and never gives up ground.** A failed call holds none of the EIDs it had reached for, and leaves untouched whatever the anchor already held from an earlier successful call. Acquiring an *empty* set is a successful acquisition of nothing — it must not, and does not, release what the anchor is already holding. + +**Re-acquiring a live anchor may shrink its EID set, never grow it.** A refresh keeps the EIDs still named and releases the ones dropped from the set; naming an EID the anchor does not already hold fails with `ErrLockSetWidened`. That restriction is what keeps the lockers deadlock-free. Deadlock freedom rests on every caller taking shared EIDs in one canonical order, and that order can only be imposed over the EIDs of a single call — an anchor that keeps earlier locks while waiting for new ones is holding locks outside it, so two anchors widening into each other's EIDs wait on each other indefinitely. `Audit` acquires once per anchor and releases when done, so no caller needs to widen. + +**Each backend bounds its own waiting.** A caller that passes no deadline still gets an answer: the Postgres backend stops after `acquireDeadline`, and the in-memory backend after its own `acquireDeadline`. Without a budget of its own a backend could only ever stop when the caller's context did, and could never report `ErrLockAcquireTimeout` — the signal `auditor.Service` reads to tell "already waited in full" from "worth another attempt". + +**Release always runs.** `ReleaseLocks` is idempotent, safe on an anchor that never acquired anything, and safe to `defer`. Its statement is deliberately detached from the caller's context, since the usual case is a deferred release on a context that is already cancelled and a skipped release would leave the EIDs locked against every replica until the lease TTL expired. It carries its own short internal deadline so a stuck release cannot outlive the call that issued it. + +### Error classification + +Callers act on the outcome of a failed acquisition, so both backends classify it the same way: + +| Outcome | Sentinels | Meaning | +|---------|-----------|---------| +| Another anchor holds an EID and the waiting budget ran out | `ErrLockContention` + `ErrLockAcquireTimeout` | A real conflict, already waited out in full; retrying only adds delay | +| Another anchor holds an EID, but a transient failure or a cancelled caller ended the wait | `ErrLockContention` | A real conflict, not yet waited out; a later attempt may succeed | +| The caller cancelled or ran out of time while nothing held the EIDs | neither (plain context error) | Not a conflict, and not counted as one | +| The backend's own waiting budget elapsed while nothing held the EIDs | neither (plain context error) | Not a conflict; worth retrying while the caller's context is still live | +| The anchor asked for an EID it does not already hold | `ErrLockSetWidened` | A caller error, not a conflict: every attempt reproduces it | +| The database failed | neither; the underlying error is preserved | An infrastructure fault, reported as itself rather than as contention | + +The distinction is drawn from whether an attempt actually lost a race for an EID — not from which context expired first. `acquireDeadline` defaults to a minute, so a request-scoped caller context is almost always the shorter of the two, and keying off it would mean the Postgres backend hardly ever reported contention in production. + +Only the two `ErrLockContention` rows count towards `auditor_audit_lock_conflicts_total`. The rows that are not conflicts are not counted as ones, so a graceful-shutdown cancellation or a database outage does not inflate the metric operators alert on for contention. + +Whether another attempt is worth making is read from the caller's context rather than from the error, because the two rows carrying a plain context error are indistinguishable by the error alone: the caller having given up is final, whereas the backend's own budget elapsing is exactly the transient case `auditor.lock` exists to retry. See `isRetriableLockError`. + ### Configuration Configure under `token.tms..auditor.locker` (see [Configuration](../configuration.md#optional-tokentmsauditorlocker)): @@ -96,8 +127,9 @@ token: backend: postgres # use "memory" (default) for single replica postgres: ttl: 30s - acquireBackoff: 100ms - acquireDeadline: 1m + acquireBackoff: 100ms # initial wait; grows exponentially, jittered + acquireMaxBackoff: 2s # cap on that growth + acquireDeadline: 1m # whole budget for waiting out contention heartbeat: 10s owner: # required; defaults to the FSC node ID ``` @@ -113,6 +145,12 @@ The Postgres backend creates an `eid_leases` table (prefixed per TMS persistence **Lease ownership:** each row is keyed by EID and carries the holding replica (`owner`) plus the request it was taken for (`anchor`). An acquisition may only take over an existing row when the lease has expired, or when the row is the same replica's lease for the *same* anchor — a re-acquisition, which just refreshes the deadline. Two different anchors therefore never hold the same EID at once, including two concurrent audits on a single replica; the second one is contended and retried until `acquireDeadline`. +### Waiting under contention + +`acquireDeadline` is the **entire** budget for waiting out a contended EID, and one layer does the waiting. Within it, the locker retries the atomic acquisition starting at `acquireBackoff`, doubling up to `acquireMaxBackoff`, with jitter on every sleep. The jitter matters as much as the growth: a fixed interval is identical on every replica, so contenders retry in lockstep and keep colliding, and at the defaults it also costs one round trip per interval for the whole deadline — hundreds per acquisition. + +`auditor.Service` wraps `AcquireLocks` in its own retry (`token.tms..auditor.lock`) for transient failures, but it does **not** retry an error carrying `ErrLockAcquireTimeout`: the locker reporting one means it already spent `acquireDeadline`, so calling it again would spend that deadline afresh rather than improve the odds. Worst-case blocking for one audit is therefore about `acquireDeadline`, not `maxRetries × acquireDeadline`. + ### Replica owner identity Every lease row carries an `owner` column, and each replica scopes all of its lease diff --git a/token/services/auditor/auditor.go b/token/services/auditor/auditor.go index 6975cde865..15a496c63c 100644 --- a/token/services/auditor/auditor.go +++ b/token/services/auditor/auditor.go @@ -178,7 +178,13 @@ func (a *Service) Audit(ctx context.Context, tx Transaction) (*token.InputStream // Acquire locks with retry and exponential backoff to prevent livelock logger.DebugfContext(ctx, "audit transaction [%s], acquire locks with retry", tx.ID()) if err := a.acquireLocksWithRetry(ctx, string(request.Anchor), eids); err != nil { - a.metrics.AuditLockConflicts.Add(1) + // Only a genuine conflict counts towards the conflict metric. Counting every + // failure meant a graceful-shutdown cancellation or a database outage — neither + // of which involves a second holder — inflated the one signal operators are + // told to alert on for contention. + if errors.Is(err, auditdb.ErrLockContention) { + a.metrics.AuditLockConflicts.Add(1) + } return nil, nil, err } @@ -193,6 +199,13 @@ func (a *Service) Audit(ctx context.Context, tx Transaction) (*token.InputStream // acquireLocksWithRetry attempts to acquire locks with exponential backoff and randomized jitter // to prevent livelock conditions when multiple auditors compete for the same enrollment IDs. // This implements the mitigation strategy for deadlock/livelock prevention. +// +// The locker owns the waiting policy and bounds it itself, so this loop must not +// re-run an attempt that already spent that budget: an error carrying +// ErrLockAcquireTimeout is final here. Retrying it anyway multiplied the locker's +// deadline by MaxRetries — worst case, ten minutes of blocking for a single audit +// against the Postgres backend, on top of the round trips each of those attempts +// spent polling. Context errors are final for the same reason: the caller is gone. func (a *Service) acquireLocksWithRetry(ctx context.Context, anchor string, eids []string) error { // Create a retry runner with jitter support retryRunner := utils.NewRetryRunnerWithJitter( @@ -204,11 +217,16 @@ func (a *Service) acquireLocksWithRetry(ctx context.Context, anchor string, eids a.lockConfig.JitterFactor, ) - // Use the retry runner to acquire locks - err := retryRunner.RunWithContext(ctx, func() error { - return a.auditDB.AcquireLocks(ctx, anchor, eids...) - }) + // Use the retry runner to acquire locks, stopping early on errors that another + // attempt cannot improve on. + err := retryRunner.RunWithErrorsContext(ctx, func() (bool, error) { + err := a.auditDB.AcquireLocks(ctx, anchor, eids...) + if err == nil { + return true, nil + } + return !isRetriableLockError(ctx, err), err + }) if err != nil { return errors.WithMessagef(err, "failed to acquire locks for anchor [%s]", anchor) } @@ -216,6 +234,30 @@ func (a *Service) acquireLocksWithRetry(ctx context.Context, anchor string, eids return nil } +// isRetriableLockError reports whether re-running AcquireLocks stands a chance of +// a different outcome. Most failures do — a contended lock may be free by now, a +// database blip may have passed — but three do not: ErrLockAcquireTimeout means +// the locker already spent its whole waiting budget, so an identical attempt would +// just spend it again; ErrLockSetWidened is a caller error that every attempt will +// reproduce; and a caller whose own context is done is no longer waiting for an +// answer. +// +// Whether the caller is gone is read from ctx, not from the error. The lockers +// bound their own waiting, and a budget of their own that elapses with nothing +// contending surfaces as a bare context.DeadlineExceeded — indistinguishable, by +// the error alone, from the caller's deadline elapsing. Classifying it from the +// error stopped the auditor after a single attempt at exactly the transient +// database failures this retry exists to survive, while ctx was still perfectly +// live. +func isRetriableLockError(ctx context.Context, err error) bool { + if ctx.Err() != nil { + return false + } + + return !errors.Is(err, auditdb.ErrLockAcquireTimeout) && + !errors.Is(err, auditdb.ErrLockSetWidened) +} + // Append adds the passed transaction to the auditor database, reusing the // audit record computed by Audit, when available. // It also releases the locks acquired by Audit. diff --git a/token/services/auditor/auditor_test.go b/token/services/auditor/auditor_test.go index 7aee3fbf3c..d25c5122f4 100644 --- a/token/services/auditor/auditor_test.go +++ b/token/services/auditor/auditor_test.go @@ -14,6 +14,8 @@ import ( "time" "github.com/LFDT-Panurus/panurus/token" + commondrivermock "github.com/LFDT-Panurus/panurus/token/core/common/driver/mock" + "github.com/LFDT-Panurus/panurus/token/core/common/metrics" drivermock "github.com/LFDT-Panurus/panurus/token/driver/mock" tokenmock "github.com/LFDT-Panurus/panurus/token/mock" "github.com/LFDT-Panurus/panurus/token/services/auditor" @@ -1000,26 +1002,7 @@ func newMockAuditLocker(acquireFunc func(ctx context.Context, anchor string, eID func newTestServiceWithMockLocker(t *testing.T, mockLocker *mockAuditLocker) *auditor.Service { t.Helper() - // Create a real store service with our mock locker - fakeStore := newFakeStore() - storeService, err := auditdb.NewStoreService(fakeStore, auditdb.WithLocker(mockLocker)) - require.NoError(t, err) - - tmsProv := &depmock.TokenManagementServiceProvider{} - tmsProv.TokenManagementServiceReturns(tmsWithExtensions{newTestManagementService(t)}, nil) - - // Create the auditor service with the store that uses our mock locker - return auditor.NewService( - token.TMSID{}, - nil, // networkProvider - storeService, - nil, // tokenDB - tmsProv, - nil, // finalityTracer - nil, // metricsProvider - nil, // checkService - nil, // lockConfig (uses defaults) - ) + return newTestServiceWithMockLockerAndMetrics(t, mockLocker, nil) } func TestService_AcquireLocksWithRetry_Success_FirstAttempt(t *testing.T) { @@ -1081,6 +1064,32 @@ func TestService_AcquireLocksWithRetry_Failure_MaxRetriesExceeded(t *testing.T) assert.Equal(t, 10, mockLocker.GetCallCount(), "Should retry max times (default is 10)") } +// TestService_AcquireLocksWithRetry_NoRetryAfterLockerTimeout covers issue +// #2040's third finding. A locker that reports ErrLockAcquireTimeout has already +// spent its own acquisition deadline waiting out the contention, so repeating the +// call spends that deadline again rather than giving the caller a fresh chance. +// This loop used to retry it MaxRetries times regardless: nested inside the +// Postgres locker's one-minute deadline that meant up to ten minutes of blocking +// for a single audit, and thousands of database round trips. +func TestService_AcquireLocksWithRetry_NoRetryAfterLockerTimeout(t *testing.T) { + mockLocker := newMockAuditLocker(func(ctx context.Context, anchor string, eIDs ...string) error { + return errors.Join(auditdb.ErrLockAcquireTimeout, auditdb.ErrLockContention) + }) + svc := newTestServiceWithMockLocker(t, mockLocker) + + _, _, err := svc.Audit(context.Background(), &auditmock.Transaction{ + IDStub: func() string { return "tx-lock-timeout" }, + RequestStub: func() *token.Request { + return token.NewRequest(newTestManagementService(t), token.RequestAnchor("tx-lock-timeout")) + }, + }) + + require.Error(t, err) + require.ErrorIs(t, err, auditdb.ErrLockAcquireTimeout, "the locker's verdict must reach the caller intact") + assert.Equal(t, 1, mockLocker.GetCallCount(), + "a locker that already exhausted its acquire deadline must not be called again") +} + func TestService_AcquireLocksWithRetry_ContextCancelled_BeforeRetry(t *testing.T) { // Mock locker that always fails mockLocker := newMockAuditLocker(func(ctx context.Context, anchor string, eIDs ...string) error { @@ -1196,3 +1205,190 @@ func TestService_AcquireLocksWithRetry_EmptyEnrollmentIDs(t *testing.T) { require.NoError(t, err) assert.Equal(t, 1, mockLocker.GetCallCount()) } + +// newTestServiceWithMockLockerAndMetrics builds a service over a real store +// service wired to mockLocker, with a metrics provider the caller can read back. +// Pass a nil provider for the cases that do not inspect metrics. +func newTestServiceWithMockLockerAndMetrics( + t *testing.T, mockLocker *mockAuditLocker, mp metrics.Provider, +) *auditor.Service { + t.Helper() + + fakeStore := newFakeStore() + storeService, err := auditdb.NewStoreService(fakeStore, auditdb.WithLocker(mockLocker)) + require.NoError(t, err) + + // Audit binds the provider-resolved TMS before it reaches the locks, so the + // provider has to be a working one even for cases only interested in locking. + tmsProv := &depmock.TokenManagementServiceProvider{} + tmsProv.TokenManagementServiceReturns(tmsWithExtensions{newTestManagementService(t)}, nil) + + return auditor.NewService( + token.TMSID{}, + nil, // networkProvider + storeService, + nil, // tokenDB + tmsProv, + nil, // finalityTracer + mp, + nil, // checkService + nil, // lockConfig (uses defaults) + ) +} + +// countingCounter records the total added, so a test can assert on whether a +// metric was touched at all. +type countingCounter struct { + mu sync.Mutex + total float64 +} + +func (c *countingCounter) With(...string) metrics.Counter { return c } + +func (c *countingCounter) Add(delta float64) { + c.mu.Lock() + defer c.mu.Unlock() + c.total += delta +} + +func (c *countingCounter) Total() float64 { + c.mu.Lock() + defer c.mu.Unlock() + + return c.total +} + +// lockConflictProvider hands out countingCounter for the lock-conflict metric and +// discards everything else. +func lockConflictProvider() (metrics.Provider, *countingCounter) { + conflicts := &countingCounter{} + mp := &commondrivermock.MetricsProvider{} + mp.NewCounterStub = func(opts metrics.CounterOpts) metrics.Counter { + if opts.Name == "auditor_audit_lock_conflicts_total" { + return conflicts + } + + return &countingCounter{} + } + mp.NewHistogramStub = func(metrics.HistogramOpts) metrics.Histogram { return discardHistogram{} } + mp.NewGaugeStub = func(metrics.GaugeOpts) metrics.Gauge { return discardGauge{} } + + return mp, conflicts +} + +type discardHistogram struct{} + +func (discardHistogram) With(...string) metrics.Histogram { return discardHistogram{} } +func (discardHistogram) Observe(float64) {} + +type discardGauge struct{} + +func (discardGauge) With(...string) metrics.Gauge { return discardGauge{} } +func (discardGauge) Add(float64) {} +func (discardGauge) Set(float64) {} + +// TestService_AcquireLocksWithRetry_RetriesTransientFailureWithLiveCaller covers +// the classification of a failure that carries a context error but did not come +// from the caller giving up. When a locker's own acquisition budget elapses with +// nothing contending, it returns a bare context.DeadlineExceeded and no sentinel — +// see the Postgres backend's default branch. Deciding from the error alone made +// that final, so the auditor stopped after one attempt at exactly the transient +// database failures this retry exists to survive, while its own context was still +// perfectly live. Whether the caller is gone is a property of ctx, not of the +// error. +func TestService_AcquireLocksWithRetry_RetriesTransientFailureWithLiveCaller(t *testing.T) { + attempts := 0 + mockLocker := newMockAuditLocker(func(context.Context, string, ...string) error { + attempts++ + if attempts < 3 { + return errors.Wrap(context.DeadlineExceeded, "acquire eid leases") + } + + return nil + }) + svc := newTestServiceWithMockLocker(t, mockLocker) + + _, _, err := svc.Audit(context.Background(), &auditmock.Transaction{ + IDStub: func() string { return "tx-lock-transient" }, + RequestStub: func() *token.Request { + return token.NewRequest(newTestManagementService(t), token.RequestAnchor("tx-lock-transient")) + }, + }) + + require.NoError(t, err) + assert.Equal(t, 3, mockLocker.GetCallCount(), + "a locker whose own budget elapsed must be retried while the caller is still waiting") +} + +// TestService_AcquireLocksWithRetry_StopsWhenCallerIsGone is the other half: once +// the caller's context is done there is nobody left to hand the locks to, so the +// loop must stop regardless of what the error says. +func TestService_AcquireLocksWithRetry_StopsWhenCallerIsGone(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + // Contention is the most retriable failure there is, so this pins the decision on + // the caller having gone rather than on the error. The cancellation happens + // inside the first attempt, so the retry loop does reach the locker once. + mockLocker := newMockAuditLocker(func(context.Context, string, ...string) error { + cancel() + + return auditdb.ErrLockContention + }) + svc := newTestServiceWithMockLocker(t, mockLocker) + + _, _, err := svc.Audit(ctx, &auditmock.Transaction{ + IDStub: func() string { return "tx-lock-caller-gone" }, + RequestStub: func() *token.Request { + return token.NewRequest(newTestManagementService(t), token.RequestAnchor("tx-lock-caller-gone")) + }, + }) + + require.Error(t, err) + assert.Equal(t, 1, mockLocker.GetCallCount(), "a cancelled caller must not be retried for") +} + +// TestService_Audit_CountsOnlyRealLockConflicts pins the lock-conflict metric to +// what the error-classification table in docs/services/auditor.md promises: a +// failure with no second holder involved is "not a conflict, and not counted as +// one". The counter was incremented for every acquisition failure, so +// graceful-shutdown cancellations and database outages inflated the one signal +// operators are told to alert on for contention. +func TestService_Audit_CountsOnlyRealLockConflicts(t *testing.T) { + t.Run("contention is counted", func(t *testing.T) { + mp, conflicts := lockConflictProvider() + mockLocker := newMockAuditLocker(func(context.Context, string, ...string) error { + return errors.Join(auditdb.ErrLockContention, auditdb.ErrLockAcquireTimeout) + }) + svc := newTestServiceWithMockLockerAndMetrics(t, mockLocker, mp) + + _, _, err := svc.Audit(context.Background(), &auditmock.Transaction{ + IDStub: func() string { return "tx-conflict" }, + RequestStub: func() *token.Request { + return token.NewRequest(newTestManagementService(t), token.RequestAnchor("tx-conflict")) + }, + }) + + require.Error(t, err) + assert.InDelta(t, 1, conflicts.Total(), 0) + }) + + t.Run("a cancelled caller is not counted", func(t *testing.T) { + mp, conflicts := lockConflictProvider() + mockLocker := newMockAuditLocker(func(ctx context.Context, _ string, _ ...string) error { + return ctx.Err() + }) + svc := newTestServiceWithMockLockerAndMetrics(t, mockLocker, mp) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _, err := svc.Audit(ctx, &auditmock.Transaction{ + IDStub: func() string { return "tx-cancelled" }, + RequestStub: func() *token.Request { + return token.NewRequest(newTestManagementService(t), token.RequestAnchor("tx-cancelled")) + }, + }) + + require.Error(t, err) + assert.Zero(t, conflicts.Total(), + "nothing held the enrollment IDs, so the failure is not a lock conflict") + }) +} diff --git a/token/services/auditor/metrics.go b/token/services/auditor/metrics.go index a5a968098e..fd756f0c03 100644 --- a/token/services/auditor/metrics.go +++ b/token/services/auditor/metrics.go @@ -16,8 +16,11 @@ type Metrics struct { // invocation (lock acquisition included), in seconds. AuditDuration metrics.Histogram - // AuditLockConflicts counts calls to Audit() that failed because - // AcquireLocks returned an error (e.g. contention or timeout). + // AuditLockConflicts counts calls to Audit() that failed because another + // anchor held one of the enrollment IDs, i.e. those whose error carries + // ErrLockContention. Failures with no second holder involved — a cancelled + // caller, a database outage — are not conflicts and are deliberately not + // counted here, so that alerting on this metric measures contention only. AuditLockConflicts metrics.Counter // AppendDuration is a histogram of the total wall-clock time for each diff --git a/token/services/storage/auditdb/locker.go b/token/services/storage/auditdb/locker.go index b53f0dadba..7f385d7876 100644 --- a/token/services/storage/auditdb/locker.go +++ b/token/services/storage/auditdb/locker.go @@ -29,6 +29,7 @@ var ( ErrLockAcquireTimeout = locker.ErrLockAcquireTimeout ErrLockLost = locker.ErrLockLost ErrLockNotHeld = locker.ErrLockNotHeld + ErrLockSetWidened = locker.ErrLockSetWidened ErrLockerOwnerRequired = locker.ErrLockerOwnerRequired ) diff --git a/token/services/storage/auditdb/locker/config.go b/token/services/storage/auditdb/locker/config.go index de2a7895b6..1b45aa2e2a 100644 --- a/token/services/storage/auditdb/locker/config.go +++ b/token/services/storage/auditdb/locker/config.go @@ -9,6 +9,7 @@ package locker import ( "time" + "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/memory" lockerpostgres "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/postgres" ) @@ -24,6 +25,7 @@ const ( // It is read from the TMS configuration under the key "auditor.locker". type Config struct { Backend Backend `yaml:"backend"` + Memory memory.Config `yaml:"memory"` Postgres lockerpostgres.Config `yaml:"postgres"` } @@ -32,6 +34,9 @@ type Config struct { func DefaultConfig() Config { return Config{ Backend: BackendMemory, + Memory: memory.Config{ + AcquireDeadline: memory.DefaultAcquireDeadline, + }, Postgres: lockerpostgres.Config{ TTL: 30 * time.Second, AcquireBackoff: 100 * time.Millisecond, diff --git a/token/services/storage/auditdb/locker/conformance_test.go b/token/services/storage/auditdb/locker/conformance_test.go new file mode 100644 index 0000000000..8f0c6b3472 --- /dev/null +++ b/token/services/storage/auditdb/locker/conformance_test.go @@ -0,0 +1,437 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package locker_test + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker" + "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/errs" + "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/memory" + lockerpostgres "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/postgres" + "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/postgres" + _ "github.com/jackc/pgx/v5/stdlib" + "github.com/stretchr/testify/require" +) + +// The auditor picks its Locker from configuration, so any behavioural difference +// between the backends is a correctness difference between two deployments of the +// same code. Issue #2040 found one that way: a request with no enrollment IDs +// appended fine on the in-memory locker and failed under Postgres. Both backends +// had tests, but no test asserted the same expectation of both — so this file +// exercises the shared contract, and every case must hold for every backend. + +// boundedWait is the waiting budget the newBounded lockers are built with. It is +// short enough to assert against in a test, where the production defaults (a +// minute) are not. +const boundedWait = 200 * time.Millisecond + +type backend struct { + name string + // new returns a fresh locker, or skips the test when its dependencies are + // unavailable. + new func(t *testing.T) locker.Locker + // newBounded returns a fresh locker whose own waiting budget is boundedWait, + // for the cases that assert on the budget being spent rather than on the + // caller's context expiring. + newBounded func(t *testing.T) locker.Locker +} + +func backends() []backend { + return []backend{ + { + name: "memory", + new: func(*testing.T) locker.Locker { return memory.New() }, + newBounded: func(*testing.T) locker.Locker { + return memory.NewWithConfig(memory.Config{AcquireDeadline: boundedWait}) + }, + }, + { + name: "postgres", + new: func(t *testing.T) locker.Locker { + t.Helper() + + return newPostgresLocker(t, 30*time.Second) + }, + newBounded: func(t *testing.T) locker.Locker { + t.Helper() + + return newPostgresLocker(t, boundedWait) + }, + }, + } +} + +// newPostgresLocker starts a throwaway Postgres and returns a locker on its own +// table. Each locker gets a unique table so cases cannot see each other's leases. +func newPostgresLocker(t *testing.T, acquireDeadline time.Duration) locker.Locker { + t.Helper() + cfg := postgres.DefaultConfig(postgres.WithDBName("test-locker-conformance")) + terminate, _, err := postgres.StartPostgres(t.Context(), cfg, nil) + require.NoError(t, err) + t.Cleanup(terminate) + db, err := sql.Open("pgx", cfg.DataSource()) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + table := "test_conformance_" + sanitize(t.Name()) + _, _ = db.Exec("DROP TABLE IF EXISTS " + table) + t.Cleanup(func() { _, _ = db.Exec("DROP TABLE IF EXISTS " + table) }) + + // The default acquireDeadline passed in by backends().new is deliberately far + // longer than any context those cases pass in, because that is the production + // shape: it defaults to a minute, so a request-scoped caller context is nearly + // always the shorter of the two. A deadline shorter than the caller's would let + // this backend pass the contention case for the wrong reason — the failure would + // be attributed to a budget this locker had spent itself, hiding that a + // caller-driven timeout reported no sentinel at all. The cases that do want to + // assert on the budget ask for it explicitly, via backends().newBounded. + l, err := lockerpostgres.New(db, table, lockerpostgres.Config{ + TTL: 30 * time.Second, + Heartbeat: 10 * time.Second, + AcquireBackoff: 10 * time.Millisecond, + AcquireDeadline: acquireDeadline, + Owner: "conformance-owner", + }, stubReplicaID{id: "conformance-owner"}) + require.NoError(t, err) + + return l +} + +// sanitize turns a subtest name into a legal, lower-case SQL identifier suffix. +func sanitize(name string) string { + out := make([]rune, 0, len(name)) + for _, r := range name { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9': + out = append(out, r) + case r >= 'A' && r <= 'Z': + out = append(out, r+('a'-'A')) + default: + out = append(out, '_') + } + } + + return string(out) +} + +// TestConformance_AcquireReleaseRoundTrip is the happy path both backends must +// agree on, including that release makes the enrollment IDs available again. +func TestConformance_AcquireReleaseRoundTrip(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice", "bob")) + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) + l.ReleaseLocks(ctx, "anchor1") + + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "alice", "bob"), + "released enrollment IDs must be claimable again") + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_EmptyEnrollmentIDs is the regression test for issue #2040's +// first finding. Acquiring an empty set succeeds on both backends, and — the part +// that used to differ — the pre-write assertion afterwards must succeed too. +// StoreService.Append calls AssertLocksHeld on every write, so under Postgres +// this combination failed with "locks lost before write" for a request whose +// inputs and outputs yielded no enrollment IDs, while the same request appended +// cleanly on the in-memory locker. +func TestConformance_EmptyEnrollmentIDs(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1"), "acquiring nothing must succeed") + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1"), + "an empty enrollment-ID set must not be reported as lost locks") + + l.ReleaseLocks(ctx, "anchor1") + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) + }) + } +} + +// TestConformance_AssertLocksHeldWithoutAcquire pins the other half of that +// contract: the assertion reports locks that were lost, not locks that were never +// taken. Auditors that validate and append without calling Audit never acquire +// anything, so this must not be an error on any backend. +func TestConformance_AssertLocksHeldWithoutAcquire(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + require.NoError(t, l.AssertLocksHeld(context.Background(), "never-acquired")) + }) + } +} + +// TestConformance_ReleaseIsIdempotent covers the deferred-release pattern the +// auditor documents: Release runs even on paths that never acquired, and may run +// twice. +func TestConformance_ReleaseIsIdempotent(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + + l.ReleaseLocks(ctx, "never-acquired") + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice")) + l.ReleaseLocks(ctx, "anchor1") + l.ReleaseLocks(ctx, "anchor1") + + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "alice")) + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_ReacquireSameAnchorIsIdempotent covers re-acquisition, which +// the Postgres locker treats as a lease refresh. The in-memory locker used to +// deadlock against its own permits here, and to leak the permits of any +// enrollment ID dropped from the anchor's set. +func TestConformance_ReacquireSameAnchorIsIdempotent(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice")) + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice"), + "re-acquiring the same set under the same anchor must not block") + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) + + l.ReleaseLocks(ctx, "anchor1") + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "alice"), + "release after a re-acquisition must leave nothing held") + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_ContentionIsReported requires a contended acquisition to fail +// in a way callers can classify. auditor.Service inspects these sentinels to +// decide whether retrying can help, so a backend that reports contention as a +// bare context error is not interchangeable with one that does not. +// +// The caller's context here is much shorter than the Postgres backend's +// AcquireDeadline, which is the production shape and the case that used to be +// misclassified: that backend decided what to report from whichever context +// expired first, so a caller-driven timeout returned a bare +// context.DeadlineExceeded with no sentinel while the in-memory locker reported +// contention for the identical situation. +func TestConformance_ContentionIsReported(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "holder", "alice")) + t.Cleanup(func() { l.ReleaseLocks(ctx, "holder") }) + + waitCtx, cancel := context.WithTimeout(ctx, 400*time.Millisecond) + defer cancel() + err := l.AcquireLocks(waitCtx, "waiter", "alice") + + require.Error(t, err, "a lock held by another anchor must not be acquirable") + require.ErrorIs(t, err, errs.ErrLockContention) + require.ErrorIs(t, err, errs.ErrLockAcquireTimeout, + "having spent the whole waiting budget must be distinguishable from plain contention") + + // The holder keeps its lock: a failed acquisition takes nothing away. + require.NoError(t, l.AssertLocksHeld(ctx, "holder")) + }) + } +} + +// TestConformance_EmptyReacquireKeepsHeldLocks pairs with EmptyEnrollmentIDs: +// acquiring nothing succeeds, and must also leave alone whatever the anchor +// already holds. The in-memory locker used to record the empty set instead, which +// released those locks while returning nil — the caller kept believing it held +// them and another anchor could take them immediately, with no error anywhere. +// The Postgres locker returns early, so this was a divergence too. +func TestConformance_EmptyReacquireKeepsHeldLocks(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice")) + require.NoError(t, l.AcquireLocks(ctx, "anchor1"), "acquiring nothing must succeed") + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1"), "the anchor still holds alice") + + waitCtx, cancel := context.WithTimeout(ctx, 300*time.Millisecond) + defer cancel() + require.Error(t, l.AcquireLocks(waitCtx, "anchor2", "alice"), + "an empty re-acquisition must not hand alice to another anchor") + + l.ReleaseLocks(ctx, "anchor1") + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "alice"), + "releasing the anchor must still release alice") + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_CancelledCallerIsNotContention is the mirror image of +// ContentionIsReported: a failure with no other holder involved must not be +// dressed up as a lock conflict. The in-memory locker attached ErrLockContention +// to every failed acquisition, so graceful-shutdown cancellations were reported +// and counted as conflicts, while Postgres returned a plain context error. +func TestConformance_CancelledCallerIsNotContention(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + + err := l.AcquireLocks(cancelled, "anchor1", "alice") + require.Error(t, err, "a caller that has given up must not be granted locks") + require.ErrorIs(t, err, context.Canceled) + require.NotErrorIs(t, err, errs.ErrLockContention, + "nothing held alice, so the failure is the caller's own cancellation") + + // The failed call retained nothing, so alice is still free. + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "alice")) + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_DeduplicatesAndSortsEnrollmentIDs covers the ordering invariant +// both backends rely on for deadlock freedom, plus duplicate handling: a request +// naming the same enrollment ID as input and output must not self-deadlock. +func TestConformance_DeduplicatesAndSortsEnrollmentIDs(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "bob", "alice", "bob", "alice")) + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) + l.ReleaseLocks(ctx, "anchor1") + + // Both IDs must have been released exactly once, in any order. + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "alice", "bob")) + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_NarrowingReleasesDroppedEnrollmentIDs covers the shrinking half +// of the refresh contract, and is the gap that hid a Postgres-only defect: its +// acquisition statement only ever inserted, so an ID dropped from a live anchor +// kept its lease row. Both AssertLocksHeld and the heartbeat's renewal count this +// replica's rows for the anchor and require exactly as many as the session +// recorded, so each leftover row failed both — the write below was rejected with +// "locks lost before write", the heartbeat gave up on its first tick, and once the +// TTL passed the abandoned lease became claimable while this replica still +// believed it held it. The in-memory backend released it and had a test saying so; +// this suite only covered the same-set and (then still permitted) widening cases, +// so nothing compared the two backends here. +func TestConformance_NarrowingReleasesDroppedEnrollmentIDs(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice", "bob")) + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "bob"), + "narrowing a live anchor's set must succeed") + + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1"), + "the anchor holds exactly what it last asked for, so nothing was lost") + + waitCtx, cancel := context.WithTimeout(ctx, boundedWait) + defer cancel() + require.NoError(t, l.AcquireLocks(waitCtx, "anchor2", "alice"), + "alice was dropped from anchor1, so it must have been released") + + // bob is still held by anchor1 and only becomes free on release. + require.Error(t, l.AcquireLocks(waitCtx, "anchor3", "bob")) + l.ReleaseLocks(ctx, "anchor1") + require.NoError(t, l.AcquireLocks(ctx, "anchor3", "bob")) + + l.ReleaseLocks(ctx, "anchor2") + l.ReleaseLocks(ctx, "anchor3") + }) + } +} + +// TestConformance_WideningLiveAnchorIsRejected pins the other half: a live +// anchor's set may shrink or stay the same, never grow. Deadlock freedom rests on +// every caller taking shared enrollment IDs in one canonical order, and that order +// can only be imposed over the IDs of a single call — an anchor that keeps earlier +// permits while waiting for new ones holds locks outside it, so two anchors +// widening into each other's IDs wait on each other forever. Both backends must +// refuse it, and must refuse it without disturbing what the anchor already holds. +func TestConformance_WideningLiveAnchorIsRejected(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.new(t) + ctx := context.Background() + + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice")) + err := l.AcquireLocks(ctx, "anchor1", "alice", "bob") + require.ErrorIs(t, err, errs.ErrLockSetWidened) + + // The refusal took nothing and gave nothing away: anchor1 still holds alice + // and its locks are not reported as lost, and bob was never reached for. + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) + require.NoError(t, l.AcquireLocks(ctx, "anchor2", "bob")) + + waitCtx, cancel := context.WithTimeout(ctx, boundedWait) + defer cancel() + require.Error(t, l.AcquireLocks(waitCtx, "anchor3", "alice"), + "anchor1 must still hold alice") + + l.ReleaseLocks(ctx, "anchor1") + l.ReleaseLocks(ctx, "anchor2") + }) + } +} + +// TestConformance_BackendBoundsItsOwnWait covers the requirement that an +// implementation bound its own waiting rather than relying on the caller's +// context. The in-memory locker had no budget: a caller with no deadline blocked +// forever, and no failure it produced could carry ErrLockAcquireTimeout — the +// signal auditor.Service reads to tell "already waited in full" from "worth +// another attempt". Every case above passes a timeout context, which is exactly +// what hid that, so this one deliberately does not. +func TestConformance_BackendBoundsItsOwnWait(t *testing.T) { + for _, b := range backends() { + t.Run(b.name, func(t *testing.T) { + l := b.newBounded(t) + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "holder", "alice")) + t.Cleanup(func() { l.ReleaseLocks(ctx, "holder") }) + + done := make(chan error, 1) + go func() { done <- l.AcquireLocks(ctx, "waiter", "alice") }() + + select { + case err := <-done: + require.ErrorIs(t, err, errs.ErrLockContention) + require.ErrorIs(t, err, errs.ErrLockAcquireTimeout, + "a backend that spent its whole budget must say so") + case <-time.After(30 * time.Second): + t.Fatal("AcquireLocks did not bound its own wait for a caller with no deadline") + } + + require.NoError(t, l.AssertLocksHeld(ctx, "holder"), "the holder keeps its lock") + }) + } +} diff --git a/token/services/storage/auditdb/locker/dedup/dedup.go b/token/services/storage/auditdb/locker/dedup/dedup.go index ec531d925e..64f38e7810 100644 --- a/token/services/storage/auditdb/locker/dedup/dedup.go +++ b/token/services/storage/auditdb/locker/dedup/dedup.go @@ -26,3 +26,54 @@ func AndSort(source []string) []string { return slice } + +// Added returns the members of want that do not already appear in held, keeping +// want's order. +// +// Lockers use it to reconcile a re-acquisition against the set an anchor already +// holds. A non-empty result over a non-empty held set is a widening: the caller +// is asking to wait for new IDs while keeping the ones it holds, which is the +// hold-and-wait the sorted ordering above cannot protect against, because the +// held IDs were ordered against an earlier call's set rather than this one's. +func Added(want, held []string) []string { + set := setOf(held) + added := make([]string, 0, len(want)) + for _, id := range want { + if _, ok := set[id]; !ok { + added = append(added, id) + } + } + + return added +} + +// Dropped returns the members of held that no longer appear in want, keeping +// held's order. +// +// These are the locks a narrowing re-acquisition must give up. Left in place they +// are unreachable, since the caller's only handle on them was the record the +// re-acquisition replaced. +func Dropped(held, want []string) []string { + if len(held) == 0 { + return nil + } + set := setOf(want) + dropped := make([]string, 0, len(held)) + for _, id := range held { + if _, ok := set[id]; !ok { + dropped = append(dropped, id) + } + } + + return dropped +} + +// setOf returns ids as a set for membership tests. +func setOf(ids []string) map[string]struct{} { + set := make(map[string]struct{}, len(ids)) + for _, id := range ids { + set[id] = struct{}{} + } + + return set +} diff --git a/token/services/storage/auditdb/locker/dedup/dedup_test.go b/token/services/storage/auditdb/locker/dedup/dedup_test.go index 1a70d205ad..0990b244b7 100644 --- a/token/services/storage/auditdb/locker/dedup/dedup_test.go +++ b/token/services/storage/auditdb/locker/dedup/dedup_test.go @@ -73,3 +73,50 @@ func TestAndSort_CanonicalOrderIsStable(t *testing.T) { assert.Equal(t, a, b) assert.Equal(t, a, c) } + +// TestAdded covers the reconciliation input the lockers refuse a widening on: a +// non-empty result over a non-empty held set means the caller is asking to take +// enrollment IDs on top of the ones it already holds. +func TestAdded(t *testing.T) { + tests := []struct { + name string + want []string + held []string + expected []string + }{ + {name: "nothing held", want: []string{"alice", "bob"}, held: nil, expected: []string{"alice", "bob"}}, + {name: "same set", want: []string{"alice", "bob"}, held: []string{"alice", "bob"}, expected: []string{}}, + {name: "narrowed", want: []string{"bob"}, held: []string{"alice", "bob"}, expected: []string{}}, + {name: "widened", want: []string{"alice", "bob"}, held: []string{"alice"}, expected: []string{"bob"}}, + {name: "replaced", want: []string{"carol"}, held: []string{"alice"}, expected: []string{"carol"}}, + {name: "nothing wanted", want: nil, held: []string{"alice"}, expected: []string{}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expected, Added(test.want, test.held)) + }) + } +} + +// TestDropped covers the other half: the locks a narrowing re-acquisition has to +// give up, which are unreachable if it does not. +func TestDropped(t *testing.T) { + tests := []struct { + name string + held []string + want []string + expected []string + }{ + {name: "nothing held", held: nil, want: []string{"alice"}, expected: nil}, + {name: "same set", held: []string{"alice", "bob"}, want: []string{"alice", "bob"}, expected: []string{}}, + {name: "narrowed", held: []string{"alice", "bob"}, want: []string{"bob"}, expected: []string{"alice"}}, + {name: "widened", held: []string{"alice"}, want: []string{"alice", "bob"}, expected: []string{}}, + {name: "replaced", held: []string{"alice"}, want: []string{"carol"}, expected: []string{"alice"}}, + {name: "nothing wanted", held: []string{"alice", "bob"}, want: nil, expected: []string{"alice", "bob"}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expected, Dropped(test.held, test.want)) + }) + } +} diff --git a/token/services/storage/auditdb/locker/errs/errors.go b/token/services/storage/auditdb/locker/errs/errors.go index 31a57b0dba..29fc2edabb 100644 --- a/token/services/storage/auditdb/locker/errs/errors.go +++ b/token/services/storage/auditdb/locker/errs/errors.go @@ -13,6 +13,13 @@ var ( ErrLockAcquireTimeout = errors.New("auditor enrollment id lock acquire timeout") ErrLockLost = errors.New("auditor enrollment id lock lost") ErrLockNotHeld = errors.New("auditor enrollment id locks not held") + // ErrLockSetWidened signals an attempt to add enrollment IDs to an anchor that + // already holds some. Acquiring the extra IDs would mean waiting for them while + // keeping the ones already held, and the already-held ones are outside the + // sorted acquisition order that makes the lockers deadlock-free — two anchors + // widening into each other's IDs form a cycle neither can break. Callers + // acquire once per anchor and release when done, so nothing needs it. + ErrLockSetWidened = errors.New("auditor enrollment id lock set cannot grow under a live anchor") // ErrLockerOwnerRequired signals that a distributed locker was configured // without a usable owner identity. The owner identifies the replica holding // each lease, so an empty value shared by every replica would make all diff --git a/token/services/storage/auditdb/locker/factory.go b/token/services/storage/auditdb/locker/factory.go index 4cc5c677ca..ac905bae3a 100644 --- a/token/services/storage/auditdb/locker/factory.go +++ b/token/services/storage/auditdb/locker/factory.go @@ -42,7 +42,7 @@ func NewFromConfig(cfg Config, store any, replicaID id.ReplicaIDProvider) (Locke replicaID, ) case BackendMemory, "": - return memory.New(), nil + return memory.NewWithConfig(cfg.Memory), nil default: return nil, errors.Errorf("unknown locker backend: %s", cfg.Backend) } diff --git a/token/services/storage/auditdb/locker/locker.go b/token/services/storage/auditdb/locker/locker.go index 236a2fdf92..55e3056e7a 100644 --- a/token/services/storage/auditdb/locker/locker.go +++ b/token/services/storage/auditdb/locker/locker.go @@ -14,9 +14,54 @@ import ( ) // Locker coordinates exclusive access to enrollment IDs during auditor processing. +// +// Implementations must be interchangeable: the auditor picks one from +// configuration, so any behavioural difference between them turns into a +// correctness difference between two deployments running the same code. The +// method contracts below are therefore normative, not descriptive. type Locker interface { + // AcquireLocks blocks until it holds every enrollment ID in eIDs on behalf of + // anchor, or fails without holding any of them. An empty eIDs set is a + // successful acquisition of nothing. + // + // Re-acquiring under a live anchor is a refresh, and the set may only shrink or + // stay the same: the IDs still named are kept, and the ones dropped from the set + // are released. Naming an ID the anchor does not already hold returns + // ErrLockSetWidened. That restriction is what keeps the lockers deadlock-free. + // Deadlock freedom rests on every caller taking shared IDs in one canonical + // order, and that order can only be imposed over the IDs of a single call — an + // anchor that keeps earlier permits while waiting for new ones holds locks + // outside it, so two anchors widening into each other's IDs wait on each other + // forever. Callers acquire once per anchor and release when done, so no caller + // needs to widen. + // + // Implementations own the waiting policy — how long, and how often they + // retry — and must bound it, so that a caller which passes no deadline of its + // own still gets an answer. A failure caused by another holder wraps + // ErrLockContention, and additionally ErrLockAcquireTimeout once the + // implementation has spent its whole waiting budget, which tells the caller + // that repeating the call adds delay rather than a fresh chance. A failure that + // spent no budget — the implementation's own deadline elapsing while nothing + // held the IDs, or a database error — carries neither, and may be worth + // retrying while the caller's context is still live. AcquireLocks(ctx context.Context, anchor string, eIDs ...string) error + + // ReleaseLocks releases everything held under anchor. It is idempotent and + // silent about unknown anchors, so it is safe to defer. ReleaseLocks(ctx context.Context, anchor string) + + // AssertLocksHeld reports whether the locks taken for anchor are still intact, + // for callers to check before committing work that assumed exclusivity. It + // returns ErrLockNotHeld when a lock this locker granted for anchor has since + // been lost — a lease that expired and was taken over by another replica. + // + // It detects lost locks, not absent ones: an anchor that holds no locks + // succeeds, whether because its enrollment-ID set was empty or because the + // caller never locked anything for it. Callers append records for anchors they + // never locked (an auditor that validates and approves without calling Audit), + // so treating that as a failure would reject legitimate writes — and would do + // so only on the backends able to notice, which is the divergence this + // contract exists to prevent. AssertLocksHeld(ctx context.Context, anchor string) error } @@ -28,5 +73,6 @@ var ( ErrLockAcquireTimeout = errs.ErrLockAcquireTimeout ErrLockLost = errs.ErrLockLost ErrLockNotHeld = errs.ErrLockNotHeld + ErrLockSetWidened = errs.ErrLockSetWidened ErrLockerOwnerRequired = errs.ErrLockerOwnerRequired ) diff --git a/token/services/storage/auditdb/locker/memory/memory.go b/token/services/storage/auditdb/locker/memory/memory.go index fbf114e56f..61019269cb 100644 --- a/token/services/storage/auditdb/locker/memory/memory.go +++ b/token/services/storage/auditdb/locker/memory/memory.go @@ -9,8 +9,10 @@ package memory import ( "context" "sync" + "time" "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/dedup" + "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/errs" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" "golang.org/x/sync/semaphore" ) @@ -18,13 +20,86 @@ import ( // Locker is the default in-memory Locker. It uses weighted semaphores // (weight 1) so that AcquireLocks respects context cancellation and deadlines. // Suitable for single-replica deployments. +// +// Enrollment-ID semaphores and per-anchor bookkeeping are held in two separate +// maps on purpose. Both are keyed by unconstrained strings of unrelated +// provenance — a request anchor and an identity's enrollment ID — so a single +// shared map would let one namespace's key resolve to the other's value type, +// and the type assertions on the way out would panic. Two maps make that +// impossible by construction rather than by convention. type Locker struct { - locks sync.Map + cfg Config + + // sems maps an enrollment ID to the weight-1 semaphore guarding it. Entries + // are never removed: the semaphore *is* the identity of the lock, so + // discarding one while a caller is blocked on it would hand the next caller a + // different semaphore and let both believe they hold the ID. + sems sync.Map + + // mu guards the anchors map itself. It is never held across a semaphore + // acquisition, so it cannot be the lock a blocked caller is waiting behind. + mu sync.Mutex + // anchors maps a request anchor to its lock bookkeeping. + anchors map[string]*anchorState +} + +// anchorState is the bookkeeping for a single anchor. Its mutex serialises the +// whole of AcquireLocks and ReleaseLocks for that anchor, which is what makes +// the reconciliation in AcquireLocks atomic: the currently held set is read, the +// difference against the requested set is acquired, and the result is written +// back with no other call for the same anchor interleaving. +// +// The lock is per anchor rather than one lock for the whole Locker because +// AcquireLocks blocks on semaphores while holding it, and the permit it waits +// for is released by some *other* anchor's ReleaseLocks. Under a single shared +// lock that release would wait for the blocked acquisition, which waits for the +// release — a deadlock. Per-anchor locks also leave acquisitions for unrelated +// anchors fully concurrent, which is the common case. +type anchorState struct { + mu sync.Mutex + // refs counts the callers that have looked this state up and not yet finished + // with it, so it cannot be evicted from under them. + refs int + // eIDs are the enrollment IDs currently held for the anchor, deduplicated and + // sorted. Written only by the caller holding mu (or, once refs drops to zero, + // by the last caller to let go of it). + eIDs []string +} + +// DefaultAcquireDeadline is the waiting budget a Locker uses when its Config +// leaves AcquireDeadline unset. It matches the Postgres backend's default so a +// deployment that switches backends does not silently change how long an audit +// can block on a contended enrollment ID. +const DefaultAcquireDeadline = time.Minute + +// Config configures the in-memory Locker. +type Config struct { + // AcquireDeadline bounds how long one AcquireLocks call waits for enrollment + // IDs held by another anchor. The Locker contract requires implementations to + // bound their own waiting: without a budget of its own this backend could only + // ever stop when the caller's context did, so a caller with no deadline blocked + // forever and no failure could be reported as ErrLockAcquireTimeout — the very + // signal callers use to tell "already waited in full" from "worth retrying". + AcquireDeadline time.Duration `yaml:"acquireDeadline"` } -// New returns an empty in-memory Locker ready for use. +// withDefaults returns c with unset or nonsensical values replaced by defaults. +func (c Config) withDefaults() Config { + if c.AcquireDeadline <= 0 { + c.AcquireDeadline = DefaultAcquireDeadline + } + + return c +} + +// New returns an empty in-memory Locker with the default configuration. func New() *Locker { - return &Locker{} + return NewWithConfig(Config{}) +} + +// NewWithConfig returns an empty in-memory Locker ready for use. +func NewWithConfig(cfg Config) *Locker { + return &Locker{cfg: cfg.withDefaults(), anchors: make(map[string]*anchorState)} } // AcquireLocks blocks until it holds the lock for every enrollment ID in eIDs, @@ -33,57 +108,229 @@ func New() *Locker { // Implementation: the enrollment IDs are deduplicated and sorted (see // dedup.AndSort) so all callers acquire shared locks in the same order and // cannot deadlock. For each ID it lazily creates a weight-1 semaphore in the -// locks map and acquires it; using semaphore.Acquire (rather than a plain -// Mutex) means a blocked acquisition still honours ctx cancellation/deadline. -// If any acquisition fails, the locks taken so far in this call are released -// and the error is returned, so the call is all-or-nothing. On success the -// sorted ID list is stored under anchor so ReleaseLocks can find it. +// sems map and acquires it; using semaphore.Acquire (rather than a plain Mutex) +// means a blocked acquisition still honours ctx cancellation/deadline. If any +// acquisition fails, the locks taken so far in this call are released and the +// error is returned, so the call is all-or-nothing — and anything the anchor +// already held from an earlier call is left untouched. +// +// The wait is bounded by cfg.AcquireDeadline as well as by ctx, so a caller with +// no deadline of its own still gets an answer, reported as ErrLockAcquireTimeout +// once that budget is spent. +// +// Re-acquiring under an anchor that is still live keeps the IDs it already holds +// and releases the ones dropped from the set, matching the distributed locker's +// lease-refresh behaviour. Without that reconciliation the previous list would +// simply be overwritten, permanently leaking the permits it recorded. The +// reconciliation runs under the anchor's lock because it is a read-modify-write +// of that record: performed lock-free, two concurrent re-acquisitions that both +// narrow the set read the same stale record, both conclude the same ID is now +// stale, and both release its permit — which panics the process, since releasing +// a semaphore more than it was acquired is a programming error in +// golang.org/x/sync/semaphore rather than a no-op. +// +// Adding IDs to a live anchor is refused with ErrLockSetWidened. The sorted order +// only covers the IDs taken within one call, so an anchor that keeps its existing +// permits while waiting for new ones is holding locks outside that order: two +// anchors widening into each other's IDs deadlock, and permanently, since this +// method holds the anchor's lock across the blocking acquisition and so blocks +// the ReleaseLocks that would break the cycle. Refusing it makes the cycle +// unconstructible rather than merely bounded. func (m *Locker) AcquireLocks(ctx context.Context, anchor string, eIDs ...string) error { deduped := dedup.AndSort(eIDs) - acquired := make([]string, 0, len(deduped)) - for _, id := range deduped { - sem, _ := m.locks.LoadOrStore(id, semaphore.NewWeighted(1)) - if err := sem.(*semaphore.Weighted).Acquire(ctx, 1); err != nil { - for _, aid := range acquired { - if s, ok := m.locks.Load(aid); ok { - s.(*semaphore.Weighted).Release(1) - } - } - - return errors.Wrapf(err, "failed to acquire lock for enrollment ID [%s]", id) + st := m.lockAnchor(anchor) + defer m.unlockAnchor(anchor, st) + + if len(deduped) == 0 { + // Acquiring nothing succeeds, and must leave whatever the anchor already + // holds in place: recording the empty set here would release those locks + // while reporting success, so the caller would go on believing it still + // held them while another anchor was free to take them. The distributed + // locker returns early for the same reason. + return nil + } + + // Only the IDs the anchor does not already hold are acquired: re-acquiring one + // this very caller holds would block on its own permit. + added := dedup.Added(deduped, st.eIDs) + if len(st.eIDs) > 0 && len(added) > 0 { + return errors.Wrapf(errs.ErrLockSetWidened, + "anchor [%s] holds %v and cannot also take %v", anchor, st.eIDs, added) + } + + // The budget bounds the blocking acquisitions below, so a caller that passes no + // deadline of its own still gets an answer. + acquireCtx, cancelAcquire := context.WithTimeout(ctx, m.cfg.AcquireDeadline) + defer cancelAcquire() + + acquired := make([]string, 0, len(added)) + for _, id := range added { + if err := m.acquireOne(acquireCtx, id); err != nil { + m.releaseAll(acquired) + + return err } acquired = append(acquired, id) } - m.locks.Store(anchor, deduped) + // Commit the new set. Anything the anchor held but no longer needs would + // otherwise be unreachable, since the caller's only handle on it was the + // record just replaced. + stale := dedup.Dropped(st.eIDs, deduped) + st.eIDs = deduped + m.releaseAll(stale) return nil } // ReleaseLocks releases every enrollment-ID lock previously acquired under -// anchor. It looks up (and deletes) the sorted ID list stored by AcquireLocks -// and releases each semaphore. It is a no-op if the anchor is unknown (e.g. -// already released), so it is safe to call more than once. +// anchor. It takes the anchor's recorded ID list, clears it, and releases each +// semaphore. It is a no-op if the anchor holds nothing (e.g. already released, +// or never acquired), so it is safe to call more than once and safe to defer. func (m *Locker) ReleaseLocks(_ context.Context, anchor string) { - dedupBoxed, ok := m.locks.LoadAndDelete(anchor) + st := m.lockAnchor(anchor) + defer m.unlockAnchor(anchor, st) + + eIDs := st.eIDs + st.eIDs = nil + m.releaseAll(eIDs) +} + +// AssertLocksHeld always succeeds for the in-memory locker: locks live in this +// process's memory and cannot be lost or stolen by another replica, so there is +// nothing to re-verify. It exists to satisfy the Locker interface, whose +// distributed implementations use it to detect a lost lease — and, per that +// interface's contract, an anchor holding no locks is not a failure either. +func (m *Locker) AssertLocksHeld(_ context.Context, _ string) error { + return nil +} + +// acquireOne takes the permit for a single enrollment ID. +// +// TryAcquire is attempted before the blocking Acquire so that a failure can be +// classified exactly: a permit that was already taken is contention, whereas one +// that was free means the caller had stopped waiting. A blocking Acquire on its +// own cannot tell those apart — it reports the context error either way — which +// is why a cancelled request used to be reported as a lock conflict. The two +// calls share one admission policy (both refuse to jump an existing queue of +// waiters), so probing first does not let this caller barge ahead. +func (m *Locker) acquireOne(ctx context.Context, eID string) error { + sem := m.semaphoreFor(eID) + if sem.TryAcquire(1) { + if err := ctx.Err(); err != nil { + // The ID was free but the caller is no longer waiting for it. Holding on + // to the permit would strand it, since a caller that gets an error does + // not go on to release anything. + sem.Release(1) + + return acquireError(err, eID, false) + } + + return nil + } + if err := sem.Acquire(ctx, 1); err != nil { + return acquireError(err, eID, true) + } + + return nil +} + +// lockAnchor returns anchor's state with its lock held, creating the state on +// first use. The reference taken here keeps the state alive until the matching +// unlockAnchor, so a concurrent caller cannot evict it mid-use. +func (m *Locker) lockAnchor(anchor string) *anchorState { + m.mu.Lock() + st, ok := m.anchors[anchor] if !ok { - return + st = &anchorState{} + m.anchors[anchor] = st + } + st.refs++ + m.mu.Unlock() + st.mu.Lock() + + return st +} + +// unlockAnchor releases anchor's lock and drops the state from the map once it +// holds no enrollment IDs and no other caller references it. Anchors are request +// identifiers, so without the eviction the map would grow by one entry for every +// transaction the process ever audits. +func (m *Locker) unlockAnchor(anchor string, st *anchorState) { + // Emptiness and the reference count are read together, under both locks, and + // st.mu is only dropped afterwards. Sampling emptiness first and acting on it + // after taking m.mu let a waiter run a whole acquisition in between: it was + // already counted in refs, so it blocked on st.mu rather than creating its own + // state, recorded its enrollment IDs, and released its reference — leaving this + // caller to evict, on the strength of a now-stale flag, an anchor that held + // permits. Nothing could reach those permits afterwards, since the next + // ReleaseLocks for the anchor created a fresh state with no IDs recorded, so + // every later audit touching them blocked until its own deadline, for the + // lifetime of the process. + // + // Taking m.mu while holding st.mu is safe in this order: lockAnchor releases + // m.mu before it waits on st.mu, so no caller ever holds m.mu while blocked on + // an anchor's lock. + m.mu.Lock() + st.refs-- + if st.refs == 0 && len(st.eIDs) == 0 { + delete(m.anchors, anchor) } - deduped := dedupBoxed.([]string) - for _, id := range deduped { - lock, ok := m.locks.Load(id) + m.mu.Unlock() + st.mu.Unlock() +} + +// semaphoreFor returns the weight-1 semaphore guarding eID, creating it on +// first use. +func (m *Locker) semaphoreFor(eID string) *semaphore.Weighted { + boxed, _ := m.sems.LoadOrStore(eID, semaphore.NewWeighted(1)) + sem, ok := boxed.(*semaphore.Weighted) + if !ok { + // Unreachable: only this method writes to sems, and only semaphores. + sem = semaphore.NewWeighted(1) + } + + return sem +} + +// releaseAll releases one permit for each enrollment ID in eIDs. +func (m *Locker) releaseAll(eIDs []string) { + for _, id := range eIDs { + boxed, ok := m.sems.Load(id) if !ok { continue } - lock.(*semaphore.Weighted).Release(1) + if sem, ok := boxed.(*semaphore.Weighted); ok { + sem.Release(1) + } } } -// AssertLocksHeld always succeeds for the in-memory locker: locks live in this -// process's memory and cannot be lost or stolen by another replica, so there is -// nothing to re-verify. It exists to satisfy the Locker interface, whose -// distributed implementations use it to detect a lost lease. -func (m *Locker) AssertLocksHeld(_ context.Context, _ string) error { - return nil +// acquireError classifies a failed acquisition of eID. semaphore.Acquire only +// fails when ctx is done, which covers two situations the caller must be able to +// tell apart: the ID was held by someone else and this caller ran out of time +// waiting for it, or the caller had already given up on its own. +// +// contended says which. Only a genuine conflict attaches the shared sentinels, +// since ErrLockContention and ErrLockAcquireTimeout are what callers test to +// decide whether a lock conflict is worth retrying and whether the waiting +// budget is already spent. Attaching them unconditionally reported every +// cancellation — a node shutting down, a caller whose own deadline elapsed +// elsewhere — as a lock conflict, which is both wrong and a divergence from the +// distributed locker, which returns a plain context error in that case. +func acquireError(err error, eID string, contended bool) error { + sentinels := make([]error, 0, 3) + if contended { + sentinels = append(sentinels, errs.ErrLockContention) + if errors.Is(err, context.DeadlineExceeded) { + // The deadline elapsed while waiting out another holder, so the whole + // waiting budget went on this attempt and repeating it adds delay rather + // than a fresh chance. + sentinels = append(sentinels, errs.ErrLockAcquireTimeout) + } + } + + return errors.Wrapf(errors.Join(append(sentinels, err)...), + "failed to acquire lock for enrollment ID [%s]", eID) } diff --git a/token/services/storage/auditdb/locker/memory/memory_test.go b/token/services/storage/auditdb/locker/memory/memory_test.go index 9960d8d48e..1d371ec5d3 100644 --- a/token/services/storage/auditdb/locker/memory/memory_test.go +++ b/token/services/storage/auditdb/locker/memory/memory_test.go @@ -12,7 +12,9 @@ import ( "testing" "time" + "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/errs" "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/locker/memory" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -54,3 +56,378 @@ func TestLocker_DeadlockPrevention(t *testing.T) { t.Fatal("deadlock detected") } } + +// mustAcquireWithin fails the test if AcquireLocks has not returned within a +// short budget. Every acquisition in these tests is expected to be uncontended, +// so a hang means the locker is blocking on a permit it already holds. +func mustAcquireWithin(t *testing.T, l *memory.Locker, anchor string, eIDs ...string) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + require.NoError(t, l.AcquireLocks(ctx, anchor, eIDs...)) +} + +// TestLocker_AnchorEqualToEnrollmentID covers the key-namespace collision from +// issue #2040. Anchors and enrollment IDs are unconstrained strings from +// unrelated sources, and both used to be stored in one sync.Map: an anchor equal +// to an enrollment ID made a later lookup return the other namespace's value +// type, and the unchecked type assertion on it panicked. Every sub-case here +// panicked, released the wrong thing, or deadlocked before the two maps were +// split. +func TestLocker_AnchorEqualToEnrollmentID(t *testing.T) { + t.Run("anchor equals its own enrollment id", func(t *testing.T) { + l := memory.New() + mustAcquireWithin(t, l, "x", "x") + l.ReleaseLocks(context.Background(), "x") + + // The enrollment ID must be free again, not stranded by the release. + mustAcquireWithin(t, l, "other", "x") + }) + + t.Run("later anchor equals an earlier enrollment id", func(t *testing.T) { + l := memory.New() + ctx := context.Background() + mustAcquireWithin(t, l, "a", "x") + + // "x" is a live semaphore key; using it as an anchor used to make the next + // acquisition assert a []string as a *semaphore.Weighted and panic. + mustAcquireWithin(t, l, "x", "y") + + l.ReleaseLocks(ctx, "a") + l.ReleaseLocks(ctx, "x") + mustAcquireWithin(t, l, "fresh", "x", "y") + }) + + t.Run("release of an anchor that is only an enrollment id", func(t *testing.T) { + l := memory.New() + ctx := context.Background() + mustAcquireWithin(t, l, "a", "x") + + // "x" was never an anchor. This used to load the semaphore stored under + // "x", assert it as []string and panic — after deleting it, stranding the + // permit anchor "a" still held. + l.ReleaseLocks(ctx, "x") + + l.ReleaseLocks(ctx, "a") + mustAcquireWithin(t, l, "b", "x") + }) +} + +// TestLocker_ReleaseUnknownAnchor documents that releasing something never +// acquired is a silent no-op, so callers can defer Release unconditionally. +func TestLocker_ReleaseUnknownAnchor(t *testing.T) { + l := memory.New() + ctx := context.Background() + + l.ReleaseLocks(ctx, "never-acquired") + mustAcquireWithin(t, l, "never-acquired", "alice") + l.ReleaseLocks(ctx, "never-acquired") + l.ReleaseLocks(ctx, "never-acquired") +} + +// TestLocker_ReacquireSameAnchor covers the permit leak that shared the same +// root cause as the collision: AcquireLocks overwrote the anchor's recorded ID +// list, so any ID dropped from it could never be released again — the caller's +// only handle on it was the list just discarded. Re-acquiring the same list also +// used to deadlock against the caller's own permits. +func TestLocker_ReacquireSameAnchor(t *testing.T) { + t.Run("same set is idempotent", func(t *testing.T) { + l := memory.New() + mustAcquireWithin(t, l, "a", "alice", "bob") + mustAcquireWithin(t, l, "a", "alice", "bob") + + l.ReleaseLocks(context.Background(), "a") + mustAcquireWithin(t, l, "b", "alice", "bob") + }) + + t.Run("dropped enrollment ids are released", func(t *testing.T) { + l := memory.New() + mustAcquireWithin(t, l, "a", "alice", "bob") + mustAcquireWithin(t, l, "a", "bob") + + // "alice" is no longer recorded under "a", so it must have been released + // rather than leaked. + mustAcquireWithin(t, l, "other", "alice") + + l.ReleaseLocks(context.Background(), "a") + mustAcquireWithin(t, l, "another", "bob") + }) + + t.Run("widening a live anchor is refused", func(t *testing.T) { + l := memory.New() + ctx := context.Background() + mustAcquireWithin(t, l, "a", "alice") + + err := l.AcquireLocks(ctx, "a", "alice", "bob") + require.ErrorIs(t, err, errs.ErrLockSetWidened) + + // The refusal changed nothing: "alice" is still held under "a", and "bob" was + // never reached for. + mustAcquireWithin(t, l, "holder-bob", "bob") + l.ReleaseLocks(ctx, "a") + mustAcquireWithin(t, l, "holder-alice", "alice") + }) +} + +// TestLocker_WideningLiveAnchorsCannotDeadlock is the regression test for the +// cross-anchor cycle. dedup.AndSort only orders the enrollment IDs taken within +// one call, so an anchor that keeps its earlier permits while waiting for new ones +// holds locks outside that order. Two anchors widening into each other's IDs then +// wait on each other — and permanently, because AcquireLocks holds the anchor's +// lock across the blocking acquisition, which blocks the ReleaseLocks that would +// break the cycle. Before widening was refused, neither call below returned. +func TestLocker_WideningLiveAnchorsCannotDeadlock(t *testing.T) { + l := memory.New() + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "a", "alice")) + require.NoError(t, l.AcquireLocks(ctx, "b", "bob")) + + done := make(chan error, 2) + go func() { done <- l.AcquireLocks(ctx, "a", "alice", "bob") }() + go func() { done <- l.AcquireLocks(ctx, "b", "alice", "bob") }() + + for range 2 { + select { + case err := <-done: + require.ErrorIs(t, err, errs.ErrLockSetWidened) + case <-time.After(5 * time.Second): + t.Fatal("deadlock detected: two anchors widening into each other's enrollment IDs") + } + } + + // Both anchors kept exactly what they held, so neither ID is free and neither is + // stranded. + l.ReleaseLocks(ctx, "a") + l.ReleaseLocks(ctx, "b") + mustAcquireWithin(t, l, "c", "alice", "bob") +} + +// TestLocker_BoundsItsOwnWait covers the waiting budget the Locker contract +// requires every implementation to have. This backend had none: a blocked +// acquisition ended only when the caller's context did, so a caller with no +// deadline waited forever, and no failure could ever carry ErrLockAcquireTimeout — +// the signal auditor.Service uses to tell "already waited in full" from "worth +// another attempt". The Postgres backend bounded itself with acquireDeadline, so +// this was a divergence between two deployments of the same code. +func TestLocker_BoundsItsOwnWait(t *testing.T) { + l := memory.NewWithConfig(memory.Config{AcquireDeadline: 150 * time.Millisecond}) + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "holder", "alice")) + + done := make(chan error, 1) + // Deliberately no deadline on the caller's context: the budget under test is the + // locker's own. + go func() { done <- l.AcquireLocks(ctx, "waiter", "alice") }() + + select { + case err := <-done: + require.ErrorIs(t, err, errs.ErrLockContention) + require.ErrorIs(t, err, errs.ErrLockAcquireTimeout, + "spending the whole budget must be distinguishable from plain contention") + case <-time.After(5 * time.Second): + t.Fatal("AcquireLocks did not bound its own wait for a caller with no deadline") + } + + require.NoError(t, l.AssertLocksHeld(ctx, "holder"), "the holder keeps its lock") +} + +// TestLocker_EvictionDoesNotStrandPermits is the regression test for the stale +// emptiness flag in unlockAnchor. It read whether the anchor still held anything +// before releasing the anchor's lock, then acted on that read after taking the map +// lock — leaving room for a waiter to run a whole acquisition in between, record +// its enrollment IDs and drop its reference, so this caller evicted an anchor that +// did hold permits. Nothing could reach them afterwards: the next ReleaseLocks for +// the anchor built a fresh state with no IDs recorded and released nothing, so +// every later audit touching those IDs blocked until its own deadline for the +// lifetime of the process. +// +// The window is small, so the interleaving has to be driven repeatedly. Each round +// checks that "x" is still acquirable by another anchor once the round's holder has +// let it go. +func TestLocker_EvictionDoesNotStrandPermits(t *testing.T) { + const rounds = 20000 + l := memory.New() + ctx := context.Background() + + for i := range rounds { + require.NoError(t, l.AcquireLocks(ctx, "a", "x")) + + var wg sync.WaitGroup + wg.Add(2) + // One caller releases the anchor while another re-acquires it, so the release + // computes emptiness for a state the acquisition is about to fill. + go func() { defer wg.Done(); l.ReleaseLocks(ctx, "a") }() + go func() { defer wg.Done(); _ = l.AcquireLocks(ctx, "a", "x") }() + wg.Wait() + + l.ReleaseLocks(ctx, "a") + + // "x" is held by nobody now, so a fresh anchor must be able to take it + // immediately. A stranded permit shows up here as a timeout. + probe, cancel := context.WithTimeout(ctx, 2*time.Second) + err := l.AcquireLocks(probe, "probe", "x") + cancel() + require.NoError(t, err, "enrollment id [x] was stranded after %d round(s)", i+1) + l.ReleaseLocks(ctx, "probe") + } +} + +// TestLocker_ConcurrentReacquireDoesNotOverRelease is the regression test for the +// reconciliation race. Narrowing an anchor's enrollment-ID set means releasing the +// IDs it dropped, and computing that from a lock-free read of the anchor's record +// let concurrent callers all see the same pre-narrowing set: every one of them +// concluded the same ID was now stale and released its permit. Releasing a +// semaphore more times than it was acquired panics +// (`semaphore: released more than held`), which takes the auditor process down. +// +// The pre-fix code survives a single-threaded narrowing, so the race has to be +// driven concurrently and repeatedly to show up. +func TestLocker_ConcurrentReacquireDoesNotOverRelease(t *testing.T) { + const ( + rounds = 200 + callers = 4 + ) + for range rounds { + l := memory.New() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + require.NoError(t, l.AcquireLocks(ctx, "a", "x", "y")) + + var wg sync.WaitGroup + wg.Add(callers) + for range callers { + go func() { + defer wg.Done() + // {x, y} narrowed to {x}: "y" is stale for whichever caller commits + // first, and must be released exactly once no matter how many callers + // computed that it was stale. + _ = l.AcquireLocks(ctx, "a", "x") + }() + } + wg.Wait() + cancel() + + // "y" was released exactly once, so exactly one other anchor can take it, + // while "x" is still held by "a". + fresh := context.Background() + require.NoError(t, l.AcquireLocks(fresh, "holder-y", "y")) + l.ReleaseLocks(fresh, "a") + require.NoError(t, l.AcquireLocks(fresh, "holder-x", "x")) + } +} + +// TestLocker_EmptyReacquireKeepsHeldLocks covers the other half of the +// reconciliation contract. Acquiring an empty set is a successful acquisition of +// nothing, so it must leave the anchor's existing locks alone; recording the empty +// set instead released them while returning nil, so the caller went on believing +// it held them and a second anchor could take them straight away — exclusivity +// gone, with no error on any path. The distributed locker returns early here, so +// this was also a backend divergence. +func TestLocker_EmptyReacquireKeepsHeldLocks(t *testing.T) { + l := memory.New() + ctx := context.Background() + + mustAcquireWithin(t, l, "a", "alice") + require.NoError(t, l.AcquireLocks(ctx, "a"), "acquiring nothing must succeed") + + timeout, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + require.Error(t, l.AcquireLocks(timeout, "b", "alice"), + "an empty re-acquisition must not release what the anchor already holds") + + // Releasing "a" must still release alice, i.e. the record was not cleared. + l.ReleaseLocks(ctx, "a") + mustAcquireWithin(t, l, "b", "alice") +} + +// TestLocker_CancelledCallerIsNotContention pins the classification of a failure +// that has nothing to do with another holder. ErrLockContention was attached to +// every failed acquisition, so cancelling a request — a node shutting down, a +// caller whose deadline elapsed elsewhere — was reported as a lock conflict, and +// counted as one. It was also a fresh divergence in the opposite direction from +// the one this change set fixes: the distributed locker returns a plain context +// error in the same situation. +func TestLocker_CancelledCallerIsNotContention(t *testing.T) { + t.Run("cancelled", func(t *testing.T) { + l := memory.New() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := l.AcquireLocks(ctx, "a", "free-id") + require.Error(t, err, "a caller that has already given up must not be granted locks") + require.ErrorIs(t, err, context.Canceled) + require.NotErrorIs(t, err, errs.ErrLockContention, + "nothing held free-id, so the failure is the caller's own cancellation") + require.NotErrorIs(t, err, errs.ErrLockAcquireTimeout) + + // The permit must not have been retained by the failed call. + mustAcquireWithin(t, l, "b", "free-id") + }) + + t.Run("deadline already elapsed", func(t *testing.T) { + l := memory.New() + ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond) + defer cancel() + <-ctx.Done() + + err := l.AcquireLocks(ctx, "a", "free-id") + require.Error(t, err) + require.NotErrorIs(t, err, errs.ErrLockContention) + require.NotErrorIs(t, err, errs.ErrLockAcquireTimeout, + "a deadline that elapsed with nothing contending is not a spent waiting budget") + + mustAcquireWithin(t, l, "b", "free-id") + }) +} + +// TestLocker_AcquireContendedReportsContention pins the error classification. A +// blocked acquisition that gives up is contention, and callers test for that with +// errors.Is against the shared sentinels; the in-memory locker used to wrap only +// the context error, so every such test was false and the backends disagreed. +func TestLocker_AcquireContendedReportsContention(t *testing.T) { + l := memory.New() + require.NoError(t, l.AcquireLocks(context.Background(), "holder", "alice")) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := l.AcquireLocks(ctx, "waiter", "alice") + + require.Error(t, err) + require.ErrorIs(t, err, errs.ErrLockContention) + require.ErrorIs(t, err, errs.ErrLockAcquireTimeout, "a deadline that elapsed while waiting is a timeout") + assert.Contains(t, err.Error(), "alice", "the error must name the enrollment ID it could not take") +} + +// TestLocker_AcquireAllOrNothing verifies the rollback path: a failed +// acquisition must leave none of the IDs it had already taken held, or the +// caller — which got an error and will not call Release — would strand them. +func TestLocker_AcquireAllOrNothing(t *testing.T) { + l := memory.New() + ctx := context.Background() + // "bob" is taken by another anchor, so acquiring {alice, bob} must fail after + // having already taken "alice". + require.NoError(t, l.AcquireLocks(ctx, "holder", "bob")) + + timeout, cancel := context.WithTimeout(ctx, 20*time.Millisecond) + defer cancel() + require.Error(t, l.AcquireLocks(timeout, "waiter", "alice", "bob")) + + // "alice" must be free again. + mustAcquireWithin(t, l, "other", "alice") +} + +// TestLocker_AssertLocksHeld documents the interface contract: the in-memory +// locker cannot lose a lock, so the assertion never fails — including for an +// anchor that holds nothing, which callers rely on when they append records +// without having locked anything. +func TestLocker_AssertLocksHeld(t *testing.T) { + l := memory.New() + ctx := context.Background() + + require.NoError(t, l.AssertLocksHeld(ctx, "never-acquired")) + require.NoError(t, l.AcquireLocks(ctx, "a")) + require.NoError(t, l.AssertLocksHeld(ctx, "a"), "an empty enrollment-ID set is a successful acquisition") + require.NoError(t, l.AcquireLocks(ctx, "b", "alice")) + require.NoError(t, l.AssertLocksHeld(ctx, "b")) + l.ReleaseLocks(ctx, "b") + require.NoError(t, l.AssertLocksHeld(ctx, "b")) +} diff --git a/token/services/storage/auditdb/locker/postgres/config.go b/token/services/storage/auditdb/locker/postgres/config.go index 62fdfca8cb..1b9aaedc31 100644 --- a/token/services/storage/auditdb/locker/postgres/config.go +++ b/token/services/storage/auditdb/locker/postgres/config.go @@ -15,19 +15,46 @@ import ( ) const ( - defaultTTL = 30 * time.Second - defaultAcquireBackoff = 100 * time.Millisecond - defaultAcquireDeadline = time.Minute - defaultHeartbeat = 10 * time.Second + defaultTTL = 30 * time.Second + defaultAcquireBackoff = 100 * time.Millisecond + defaultAcquireMaxBackoff = 2 * time.Second + defaultAcquireDeadline = time.Minute + defaultHeartbeat = 10 * time.Second ) +// Backoff growth for the acquisition loop. The multiplier matches the auditor's +// own retry defaults; the jitter is what keeps contending replicas from retrying +// in lockstep, so it is deliberately not configurable to zero. +const ( + acquireBackoffMultiplier = 2.0 + acquireJitterFactor = 0.3 +) + +// releaseTimeout bounds the lease-deleting statement in releaseAnchor. That +// statement deliberately runs on a context detached from the caller's, so it +// needs a deadline of its own or it could block for as long as the connection +// pool and the lock queue make it. It is not configurable because it is not a +// tuning knob: it exists only so a stuck delete cannot outlive the call that +// issued it, and failing it costs nothing beyond leaving the leases to expire on +// their TTL, which is what a crashed replica does anyway. +const releaseTimeout = 5 * time.Second + // Config holds Postgres lease-table locking settings. type Config struct { // TTL is the lease duration for each EID lock row. TTL time.Duration `yaml:"ttl"` - // AcquireBackoff is the wait between retry attempts when a lock is contended. + // AcquireBackoff is the initial wait between retry attempts when a lock is + // contended. Successive waits grow exponentially and are jittered, so this is + // the floor rather than a fixed poll interval. AcquireBackoff time.Duration `yaml:"acquireBackoff"` - // AcquireDeadline is the total time allowed to acquire all EID locks. + // AcquireMaxBackoff caps the exponential growth of AcquireBackoff, so a long + // wait keeps checking at a steady rate instead of drifting towards the + // deadline. + AcquireMaxBackoff time.Duration `yaml:"acquireMaxBackoff"` + // AcquireDeadline is the total time allowed to acquire all EID locks. It is + // the entire budget for waiting out contention: callers are expected not to + // wrap AcquireLocks in a retry loop of their own, since that would multiply + // this deadline by their own attempt count. AcquireDeadline time.Duration `yaml:"acquireDeadline"` // Heartbeat is the interval at which held leases are renewed (~TTL/3). Heartbeat time.Duration `yaml:"heartbeat"` @@ -45,6 +72,14 @@ func (c Config) withDefaults(owner string) Config { if c.AcquireBackoff <= 0 { c.AcquireBackoff = defaultAcquireBackoff } + if c.AcquireMaxBackoff <= 0 { + c.AcquireMaxBackoff = defaultAcquireMaxBackoff + } + if c.AcquireMaxBackoff < c.AcquireBackoff { + // A cap below the floor would clamp every wait to the cap, silently + // undoing the exponential growth. + c.AcquireMaxBackoff = c.AcquireBackoff + } if c.AcquireDeadline <= 0 { c.AcquireDeadline = defaultAcquireDeadline } diff --git a/token/services/storage/auditdb/locker/postgres/config_test.go b/token/services/storage/auditdb/locker/postgres/config_test.go index b2d14c7f23..87eb2e17eb 100644 --- a/token/services/storage/auditdb/locker/postgres/config_test.go +++ b/token/services/storage/auditdb/locker/postgres/config_test.go @@ -29,30 +29,33 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: "replica-1", expectedOwner: "replica-1", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "replica-1", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "replica-1", }, }, { name: "explicit values preserved", cfg: Config{ - TTL: time.Minute, - AcquireBackoff: time.Second, - AcquireDeadline: 2 * time.Minute, - Heartbeat: 20 * time.Second, - Owner: "cfg-owner", + TTL: time.Minute, + AcquireBackoff: time.Second, + AcquireMaxBackoff: 5 * time.Second, + AcquireDeadline: 2 * time.Minute, + Heartbeat: 20 * time.Second, + Owner: "cfg-owner", }, replicaID: "replica-1", expectedOwner: "cfg-owner", expected: Config{ - TTL: time.Minute, - AcquireBackoff: time.Second, - AcquireDeadline: 2 * time.Minute, - Heartbeat: 20 * time.Second, - Owner: "cfg-owner", + TTL: time.Minute, + AcquireBackoff: time.Second, + AcquireMaxBackoff: 5 * time.Second, + AcquireDeadline: 2 * time.Minute, + Heartbeat: 20 * time.Second, + Owner: "cfg-owner", }, }, { @@ -61,11 +64,12 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: "replica-1", expectedOwner: "cfg-owner", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "cfg-owner", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "cfg-owner", }, }, { @@ -74,11 +78,12 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: "replica-1", expectedOwner: "o", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "o", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "o", }, }, { @@ -87,11 +92,12 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: "", expectedOwner: "", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "", }, }, { @@ -100,11 +106,12 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: "replica-1", expectedOwner: "replica-1", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "replica-1", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "replica-1", }, }, { @@ -113,11 +120,12 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: " ", expectedOwner: "", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "", }, }, { @@ -126,11 +134,12 @@ func TestConfig_WithDefaults(t *testing.T) { replicaID: "replica-1", expectedOwner: "cfg-owner", expected: Config{ - TTL: defaultTTL, - AcquireBackoff: defaultAcquireBackoff, - AcquireDeadline: defaultAcquireDeadline, - Heartbeat: defaultHeartbeat, - Owner: "cfg-owner", + TTL: defaultTTL, + AcquireBackoff: defaultAcquireBackoff, + AcquireMaxBackoff: defaultAcquireMaxBackoff, + AcquireDeadline: defaultAcquireDeadline, + Heartbeat: defaultHeartbeat, + Owner: "cfg-owner", }, }, } @@ -143,6 +152,36 @@ func TestConfig_WithDefaults(t *testing.T) { } } +// TestConfig_AcquireMaxBackoffClamped covers the one relationship between the two +// backoff knobs. A cap below the initial wait would clamp every attempt to the +// cap, silently cancelling the exponential growth it is supposed to bound, so the +// cap is raised to the floor instead. +func TestConfig_AcquireMaxBackoffClamped(t *testing.T) { + tests := []struct { + name string + backoff time.Duration + maxBackA time.Duration + expected time.Duration + }{ + {name: "unset cap takes the default", backoff: 10 * time.Millisecond, maxBackA: 0, expected: defaultAcquireMaxBackoff}, + {name: "cap below the floor is raised to it", backoff: 5 * time.Second, maxBackA: time.Second, expected: 5 * time.Second}, + {name: "cap above the floor is kept", backoff: 100 * time.Millisecond, maxBackA: time.Second, expected: time.Second}, + {name: "cap equal to the floor is kept", backoff: time.Second, maxBackA: time.Second, expected: time.Second}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := Config{ + AcquireBackoff: test.backoff, + AcquireMaxBackoff: test.maxBackA, + Owner: "o", + }.withDefaults("") + assert.Equal(t, test.expected, got.AcquireMaxBackoff) + assert.GreaterOrEqual(t, got.AcquireMaxBackoff, got.AcquireBackoff, + "the cap must never sit below the initial wait") + }) + } +} + func TestConfig_Validate(t *testing.T) { tests := []struct { name string diff --git a/token/services/storage/auditdb/locker/postgres/postgres.go b/token/services/storage/auditdb/locker/postgres/postgres.go index 1c9866ecc3..72371f1747 100644 --- a/token/services/storage/auditdb/locker/postgres/postgres.go +++ b/token/services/storage/auditdb/locker/postgres/postgres.go @@ -21,6 +21,7 @@ import ( q "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query" qcommon "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/common" "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/cond" + "github.com/LFDT-Panurus/panurus/token/services/utils" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" ) @@ -109,49 +110,148 @@ func (p *Locker) createSchema() error { // Implementation: the IDs are deduplicated and sorted (dedup.AndSort) for the // same deadlock-free ordering the in-memory locker relies on. It then retries // tryAcquireAll — a single atomic upsert that succeeds only if it can claim -// every ID — sleeping AcquireBackoff between attempts and giving up with a -// contention error once AcquireDeadline passes (or ctx is cancelled). On -// success it records the held IDs under anchor and starts a background -// heartbeat that renews the leases before they expire, so a long-running audit -// keeps its locks while a crashed replica's leases expire and become claimable -// by others. Any partial state is released on the give-up/cancel paths. +// every ID — until AcquireDeadline passes (or ctx is cancelled), backing off +// between attempts with exponential growth and jitter. On success it records the +// held IDs under anchor and starts a background heartbeat that renews the leases +// before they expire, so a long-running audit keeps its locks while a crashed +// replica's leases expire and become claimable by others. Any partial state is +// released on the give-up/cancel paths, except over an anchor that already holds +// a live session, whose leases are left alone. +// +// A failure caused by another holder joins ErrLockContention, and additionally +// ErrLockAcquireTimeout once the waiting budget is spent, which tells callers +// this locker already waited and the attempt should not simply be repeated (see +// auditor.Service.acquireLocksWithRetry). AcquireDeadline is that budget. +// +// An empty eIDs set is a successful acquisition of nothing: there is no lease to +// take, so no session is opened and AssertLocksHeld has nothing to verify. +// +// Re-acquiring under a live anchor may keep or shrink its set, never grow it: see +// the Locker contract for why widening is refused with ErrLockSetWidened, and +// releaseDropped for what shrinking has to clean up. func (p *Locker) AcquireLocks(ctx context.Context, anchor string, eIDs ...string) error { deduped := dedup.AndSort(eIDs) if len(deduped) == 0 { return nil } - deadline := time.Now().Add(p.cfg.AcquireDeadline) - for { - ok, err := p.tryAcquireAll(ctx, anchor, deduped) + held := p.sessionEIDs(anchor) + if added := dedup.Added(deduped, held); len(held) > 0 && len(added) > 0 { + return errors.Wrapf(errs.ErrLockSetWidened, + "anchor [%s] holds %v and cannot also take %v", anchor, held, added) + } + + // The deadline is enforced through a derived context so it also bounds the + // backoff sleeps and the queries themselves, not just the attempt loop. + acquireCtx, cancelAcquire := context.WithTimeout(ctx, p.cfg.AcquireDeadline) + defer cancelAcquire() + + // contended records whether any attempt actually lost a race for one of the + // IDs. The outcome is classified from this rather than from which context + // expired first: acquireCtx carries AcquireDeadline, a minute by default, so a + // request-scoped caller context is nearly always the shorter of the two. + // Keying off the caller's context meant that in production this backend + // reported genuine contention as a bare context error — no sentinel at all — + // while the in-memory locker reported it as contention, which is exactly the + // kind of divergence the Locker contract exists to prevent. + contended := false + err := p.acquireRunner().RunWithErrorsContext(acquireCtx, func() (bool, error) { + ok, err := p.tryAcquireAll(acquireCtx, anchor, deduped) if err != nil { - return err + return true, err } - if ok { - hbCtx, cancel := context.WithCancel(context.Background()) - p.mu.Lock() - if prev, exists := p.sessions[anchor]; exists { - prev.cancel() - } - p.sessions[anchor] = &lockSession{eIDs: deduped, cancel: cancel} - p.mu.Unlock() - go p.heartbeatLoop(hbCtx, anchor, len(deduped)) - - return nil + if !ok { + contended = true } - if time.Now().After(deadline) { - _ = p.releaseAnchor(ctx, anchor) - return errors.Join(errs.ErrLockAcquireTimeout, errs.ErrLockContention) + return ok, nil + }) + if err == nil { + // The upsert refreshed the leases named in this call, but a narrowing + // re-acquisition also has to give up the ones it dropped. + if err := p.releaseDropped(ctx, anchor, dedup.Dropped(held, deduped)); err != nil { + return err } - select { - case <-ctx.Done(): - _ = p.releaseAnchor(ctx, anchor) + p.startSession(anchor, deduped) - return ctx.Err() - case <-time.After(p.cfg.AcquireBackoff): - } + return nil + } + + // A failed acquisition must leave a session the anchor already had intact. + // tryAcquireAll rolls back unless it claims every ID, so a failure normally + // leaves no rows behind at all and this cleanup only matters for a commit that + // was applied but reported as failed. Running it over a live session would + // delete that session's lease rows while its heartbeat kept going: the next + // renewal would match nothing, report the leases lost and exit, and the + // caller's next legitimate Append would fail its pre-write assertion even + // though no lock had been stolen. + if !p.hasSession(anchor) { + _ = p.releaseAnchor(ctx, anchor) } + + // Whether the waiting budget is spent. acquireCtx reaching its deadline means + // either AcquireDeadline elapsed or the caller's own deadline did, and in both + // cases an identical retry adds delay rather than a fresh chance. A loop ended + // by a database error leaves it un-expired, and an explicitly cancelled caller + // is a cancellation rather than a timeout — matching how the in-memory locker + // classifies the same two situations. + deadlineElapsed := errors.Is(acquireCtx.Err(), context.DeadlineExceeded) + + switch { + case contended && deadlineElapsed: + // err is joined in rather than discarded: when the loop was ended by a query + // the deadline killed, it is the only record of what the database was doing. + return errors.Wrapf( + errors.Join(errs.ErrLockContention, errs.ErrLockAcquireTimeout, err), + "gave up acquiring eid leases for anchor [%s] after %v", anchor, p.cfg.AcquireDeadline) + case contended: + // The contention was real, but what ended the loop was a database failure or + // a cancelled caller, not the waiting budget. Reporting ErrLockAcquireTimeout + // here would tell the caller the budget was spent and stop it retrying a + // transient failure that a later attempt could get past. + return errors.Wrapf(errors.Join(errs.ErrLockContention, err), + "failed to acquire contended eid leases for anchor [%s]", anchor) + case ctx.Err() != nil: + return ctx.Err() + default: + return err + } +} + +// acquireRunner builds the backoff policy for the acquisition loop: exponential +// growth from AcquireBackoff, capped at AcquireMaxBackoff, with jitter. +// +// A fixed poll interval is the wrong shape here on two counts. It costs one +// round trip per interval for the whole deadline — at the defaults, hundreds per +// acquisition — and, being identical on every replica, it keeps contenders +// phase-locked so they retry in lockstep and collide again. Jittered exponential +// backoff spreads them out and cuts the round trips to a few dozen. The attempt +// count is unbounded because the derived context, not a counter, is what ends +// the loop. +func (p *Locker) acquireRunner() utils.RetryRunner { + return utils.NewRetryRunnerWithJitter( + logger, + utils.Infinitely, + p.cfg.AcquireBackoff, + p.cfg.AcquireMaxBackoff, + acquireBackoffMultiplier, + acquireJitterFactor, + ) +} + +// startSession records the leases held under anchor and starts the heartbeat that +// keeps them alive. Any session previously tracked for the anchor is cancelled +// first so its heartbeat cannot outlive it. +func (p *Locker) startSession(anchor string, eIDs []string) { + hbCtx, cancel := context.WithCancel(context.Background()) + p.mu.Lock() + if prev, exists := p.sessions[anchor]; exists { + prev.cancel() + } + p.sessions[anchor] = &lockSession{eIDs: eIDs, cancel: cancel} + p.mu.Unlock() + + go p.heartbeatLoop(hbCtx, anchor, len(eIDs)) } // tryAcquireAll attempts to claim all eIDs in a single transaction. It runs the @@ -253,21 +353,108 @@ func (p *Locker) ReleaseLocks(ctx context.Context, anchor string) { // owner so a replica only ever removes leases it still holds (never one that // expired and was since claimed by another replica), which makes it safe to // call even on the timeout/cancel paths of AcquireLocks. +// +// The delete runs on a context detached from the caller's but bounded by +// releaseTimeout. Detaching matters because the common case is a deferred release +// on a context that is already done, and a skipped delete leaves the enrollment +// IDs locked against every replica until their TTL expires. The bound matters +// just as much: a context that can neither be cancelled nor time out lets +// database/sql block indefinitely waiting for a free pooled connection or a +// conflicting row lock, which would leave AcquireLocks hanging past the very +// deadline it promises to honour, and would leak the goroutine on shutdown. func (p *Locker) releaseAnchor(ctx context.Context, anchor string) error { + releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), releaseTimeout) + defer cancel() + query, args := q.DeleteFrom(p.table). Where(cond.And(cond.Eq("anchor", anchor), cond.Eq("owner", p.cfg.Owner))). Format(p.ci) - _, err := p.db.ExecContext(ctx, query, args...) + _, err := p.db.ExecContext(releaseCtx, query, args...) return errors.Wrap(err, "release eid leases") } +// releaseDropped deletes this replica's lease rows for the enrollment IDs an +// anchor no longer needs, so that a narrowing re-acquisition gives up the leases +// it dropped instead of leaving them behind. +// +// Leaving them behind was not merely a leak. renewLeases and AssertLocksHeld both +// count this replica's un-expired rows for the anchor and require exactly as many +// as the session recorded, so every extra row made both fail: the heartbeat +// reported the leases lost on its first tick and exited, StoreService.Append +// rejected the next legitimate write with "locks lost before write", and once the +// TTL passed the abandoned leases expired and became claimable by another replica +// while this one still believed it held them. The in-memory backend released them, +// so this was a divergence between two deployments of the same code as well. +// +// A failed delete leaves the previous session untouched and reports the error. +// That state is consistent: the rows in the table are exactly the ones the old +// session recorded (a narrowing set is a subset, and the upsert only refreshed +// their expiry), so its heartbeat and assertions keep matching. Releasing the +// anchor here instead would delete the live session's rows and break precisely +// what the failure path below is careful to preserve. +func (p *Locker) releaseDropped(ctx context.Context, anchor string, eIDs []string) error { + if len(eIDs) == 0 { + return nil + } + + releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), releaseTimeout) + defer cancel() + + query, args := q.DeleteFrom(p.table). + Where(cond.And( + cond.Eq("anchor", anchor), + cond.Eq("owner", p.cfg.Owner), + cond.In[string]("eid", eIDs...), + )). + Format(p.ci) + if _, err := p.db.ExecContext(releaseCtx, query, args...); err != nil { + return errors.Wrapf(err, "failed to release eid leases dropped from anchor [%s]", anchor) + } + + return nil +} + +// sessionEIDs returns the enrollment IDs recorded for anchor, or nil when no live +// session is tracked for it. +func (p *Locker) sessionEIDs(anchor string) []string { + p.mu.Lock() + defer p.mu.Unlock() + if s, ok := p.sessions[anchor]; ok { + return s.eIDs + } + + return nil +} + +// hasSession reports whether a live lock session is tracked for anchor, i.e. +// whether an earlier AcquireLocks succeeded for it and has not been released. +func (p *Locker) hasSession(anchor string) bool { + p.mu.Lock() + defer p.mu.Unlock() + _, ok := p.sessions[anchor] + + return ok +} + // AssertLocksHeld verifies this replica still holds every lease it acquired for // anchor. It compares the number of IDs recorded locally at acquisition time // against the count of matching, non-expired, owner-scoped rows in the table. -// A mismatch (or no local record) means a lease expired and may have been -// taken over by another replica, so it returns ErrLockNotHeld. Callers use this -// after long-running work to confirm their locks were not silently lost. +// A mismatch means a lease expired and may have been taken over by another +// replica, so it returns ErrLockNotHeld. Callers use this after long-running +// work to confirm their locks were not silently lost. +// +// An anchor with no recorded session holds no leases, so there is nothing that +// could have been lost and the assertion succeeds. Reporting ErrLockNotHeld +// instead conflated "lost the locks I took" with "never took any", which failed +// two legitimate flows under this backend while both succeeded under the +// in-memory locker: a request whose inputs and outputs yield no enrollment IDs +// at all, and an auditor that validates and appends without calling Audit (so +// never acquires locks) — see the dvp and nft auditor views. +// +// A session that failed renewal is not affected: heartbeatLoop leaves the +// session in place when it gives up, so its recorded IDs are still counted here +// and the mismatch is still reported. func (p *Locker) AssertLocksHeld(ctx context.Context, anchor string) error { p.mu.Lock() s, ok := p.sessions[anchor] @@ -277,8 +464,8 @@ func (p *Locker) AssertLocksHeld(ctx context.Context, anchor string) error { } p.mu.Unlock() - if !ok || expected == 0 { - return errs.ErrLockNotHeld + if expected == 0 { + return nil } var held int diff --git a/token/services/storage/auditdb/locker/postgres/postgres_test.go b/token/services/storage/auditdb/locker/postgres/postgres_test.go index b73d97cd41..78df5521d6 100644 --- a/token/services/storage/auditdb/locker/postgres/postgres_test.go +++ b/token/services/storage/auditdb/locker/postgres/postgres_test.go @@ -202,11 +202,12 @@ func TestLocker_SameOwnerDifferentAnchorsCannotShareEID(t *testing.T) { assert.Equal(t, "owner-1", owner) require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) - // The failed attempt left nothing behind. + // The failed attempt left nothing behind. This is asserted on the table rather + // than through AssertLocksHeld, which reports lost locks and not absent ones: + // an anchor holding nothing has nothing to lose, so it succeeds by contract. var count int require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM "+table+" WHERE anchor = $1", "anchor2").Scan(&count)) assert.Equal(t, 0, count) - require.ErrorIs(t, l.AssertLocksHeld(ctx, "anchor2"), errs.ErrLockNotHeld) l.ReleaseLocks(ctx, "anchor1") @@ -328,6 +329,238 @@ func TestLocker_ConcurrentSharedEIDSingleWinner(t *testing.T) { l.ReleaseLocks(ctx, winners[0]) } +// TestLocker_NoLocksHeldAssertsSuccessfully is the regression test for the +// backend divergence in issue #2040. AssertLocksHeld reports locks that were +// lost, not locks that were never taken, so an anchor holding nothing must +// succeed. It used to return ErrLockNotHeld for any anchor without a session, +// which failed StoreService.Append with "locks lost before write" for two +// legitimate flows — a request whose inputs and outputs yield no enrollment IDs, +// and an auditor that validates and appends without calling Audit — while both +// succeeded under the in-memory locker. +func TestLocker_NoLocksHeldAssertsSuccessfully(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_nolocks" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + l := newLocker(t, db, table, lockerpostgres.Config{ + TTL: 5 * time.Second, Heartbeat: 2 * time.Second, Owner: "owner-1", + }) + ctx := context.Background() + + require.NoError(t, l.AssertLocksHeld(ctx, "never-acquired"), + "an anchor that never locked anything has nothing to lose") + + require.NoError(t, l.AcquireLocks(ctx, "empty-anchor"), "acquiring no enrollment IDs must succeed") + require.NoError(t, l.AssertLocksHeld(ctx, "empty-anchor"), + "an empty enrollment-ID set must not look like a lost lease") + + var count int + require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM "+table).Scan(&count)) + assert.Equal(t, 0, count, "an empty acquisition must not write lease rows") + + l.ReleaseLocks(ctx, "empty-anchor") + require.NoError(t, l.AssertLocksHeld(ctx, "empty-anchor")) + + // A real acquisition still tracks its leases, and losing one is still reported. + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice")) + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) + _, err := db.Exec("DELETE FROM "+table+" WHERE eid = $1", "alice") + require.NoError(t, err) + require.ErrorIs(t, l.AssertLocksHeld(ctx, "anchor1"), errs.ErrLockNotHeld, + "a lease that vanished must still be reported as lost") +} + +// TestLocker_ContentionBacksOffWithoutBusyPolling checks the shape of the retry +// loop, not just its outcome. A contended acquisition used to poll at a fixed +// AcquireBackoff for the whole deadline — hundreds of round trips per attempt, on +// an interval identical across replicas so contenders stayed in lockstep. With +// exponential backoff the same wait costs a fraction of the attempts, while the +// deadline is still honoured. +func TestLocker_ContentionBacksOffWithoutBusyPolling(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_backoff" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + l := newLocker(t, db, table, lockerpostgres.Config{ + TTL: 30 * time.Second, Heartbeat: time.Hour, Owner: "owner-1", + AcquireBackoff: 10 * time.Millisecond, + AcquireDeadline: 700 * time.Millisecond, + }) + + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "holder", "alice")) + t.Cleanup(func() { l.ReleaseLocks(ctx, "holder") }) + + start := time.Now() + err := l.AcquireLocks(ctx, "waiter", "alice") + elapsed := time.Since(start) + + require.ErrorIs(t, err, errs.ErrLockAcquireTimeout) + require.ErrorIs(t, err, errs.ErrLockContention) + assert.GreaterOrEqual(t, elapsed, 700*time.Millisecond, "the acquire deadline must be spent before giving up") + assert.Less(t, elapsed, 3*time.Second, + "the deadline bounds the whole wait, including a backoff sleep that would overrun it") + + // The failed attempt must leave the holder's lease untouched. + var anchor string + require.NoError(t, db.QueryRow("SELECT anchor FROM "+table+" WHERE eid = $1", "alice").Scan(&anchor)) + assert.Equal(t, "holder", anchor) +} + +// TestLocker_ContentionReportedWhenCallerDeadlineIsShorter pins the sentinel on +// the shape production actually has. AcquireDeadline defaults to a minute, so a +// request-scoped caller context is nearly always the shorter of the two, and the +// classification used to be decided by whichever expired first: when the caller's +// did, the contention sentinels were skipped entirely and a bare +// context.DeadlineExceeded came back. The effect was that this backend almost +// never reported contention in production — while the in-memory locker always did +// — so auditor.Service could not tell a lock conflict from a dead caller. +func TestLocker_ContentionReportedWhenCallerDeadlineIsShorter(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_shortcaller" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + l := newLocker(t, db, table, lockerpostgres.Config{ + TTL: 30 * time.Second, Heartbeat: time.Hour, Owner: "owner-1", + AcquireBackoff: 10 * time.Millisecond, + // Far longer than the caller's context below, as in a default deployment. + AcquireDeadline: 30 * time.Second, + }) + + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "holder", "alice")) + t.Cleanup(func() { l.ReleaseLocks(ctx, "holder") }) + + waitCtx, cancel := context.WithTimeout(ctx, 300*time.Millisecond) + defer cancel() + err := l.AcquireLocks(waitCtx, "waiter", "alice") + + require.Error(t, err) + require.ErrorIs(t, err, errs.ErrLockContention, + "alice was held by another anchor, which is contention however the wait ended") + require.ErrorIs(t, err, errs.ErrLockAcquireTimeout, + "the caller's budget is spent, so repeating the call adds delay rather than a fresh chance") +} + +// TestLocker_UncontendedTimeoutIsNotContention is the other side of that +// classification. A deadline that elapses while nothing holds the IDs is not a +// lock conflict, and reporting one hid the real cause: the outcome was derived +// from the error merely containing context.DeadlineExceeded, so a query the +// deadline killed on an overloaded database came back as ErrLockContention with +// the original error discarded — an infrastructure failure permanently labelled +// as contention, with no diagnostics, and one auditor.Service refuses to retry. +func TestLocker_UncontendedTimeoutIsNotContention(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_uncontended" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + l := newLocker(t, db, table, lockerpostgres.Config{ + TTL: 30 * time.Second, Heartbeat: time.Hour, Owner: "owner-1", + AcquireBackoff: time.Millisecond, + // Too short to complete a round trip, so the attempt fails on the deadline + // while "alice" is free and nothing is contending for it. + AcquireDeadline: time.Nanosecond, + }) + + err := l.AcquireLocks(context.Background(), "anchor1", "alice") + require.Error(t, err) + require.NotErrorIs(t, err, errs.ErrLockContention, + "no other anchor held alice, so this is a timeout and not a lock conflict") + + // Nothing was left behind, and the ID is still claimable. + var count int + require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM "+table).Scan(&count)) + assert.Equal(t, 0, count) +} + +// TestLocker_RejectedReacquireKeepsLiveSession covers what a re-acquisition that +// cannot be granted must leave behind. The failure path used to release the anchor +// unconditionally, so a re-acquisition that lost a race deleted the lease rows of +// the *live* session it was re-acquiring for — and left that session's record and +// heartbeat in place. The next renewal then matched none of its rows, logged the +// leases as lost and exited, and the caller's next legitimate Append failed its +// pre-write assertion with "locks lost before write" even though nothing had been +// stolen. +// +// Widening a live anchor is now refused outright (see the Locker contract), which +// is how the case below is turned away, so the release-on-failure path is reached +// only by a database error. The outcome under test is the same either way, and the +// one that matters: a re-acquisition this locker did not grant must leave the +// session's leases exactly as they were. +func TestLocker_RejectedReacquireKeepsLiveSession(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_reacquire_fail" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + cfg := lockerpostgres.Config{ + TTL: 30 * time.Second, Heartbeat: time.Hour, + AcquireBackoff: 10 * time.Millisecond, + AcquireDeadline: 300 * time.Millisecond, + } + cfg.Owner = "owner-1" + mine := newLocker(t, db, table, cfg) + cfg.Owner = "owner-2" + theirs := newLocker(t, db, table, cfg) + + ctx := context.Background() + require.NoError(t, mine.AcquireLocks(ctx, "anchor1", "alice")) + t.Cleanup(func() { mine.ReleaseLocks(ctx, "anchor1") }) + + // Another replica holds "bob", so widening anchor1 to {alice, bob} could not have + // succeeded on its merits either. + require.NoError(t, theirs.AcquireLocks(ctx, "other-anchor", "bob")) + t.Cleanup(func() { theirs.ReleaseLocks(ctx, "other-anchor") }) + + err := mine.AcquireLocks(ctx, "anchor1", "alice", "bob") + require.Error(t, err, "anchor1 already holds alice, so it cannot also take bob") + require.ErrorIs(t, err, errs.ErrLockSetWidened) + + // The live session is intact: its lease row survived and the assertion passes. + var count int + require.NoError(t, db.QueryRow( + "SELECT COUNT(*) FROM "+table+" WHERE anchor = $1 AND owner = $2", "anchor1", "owner-1").Scan(&count)) + assert.Equal(t, 1, count, "the failed re-acquisition must not delete the live session's lease") + require.NoError(t, mine.AssertLocksHeld(ctx, "anchor1"), + "a failed re-acquisition must not make the session's existing locks look lost") +} + +// TestLocker_ReleaseWithCancelledContextStillDeletes covers the common shape of a +// release: it is deferred, so by the time it runs the caller's context is often +// already done. Passing that context straight to the DELETE made the statement +// fail silently, leaving the enrollment IDs locked against every replica until +// their TTL expired — a 30-second stall by default for a lock that was released +// on time. +func TestLocker_ReleaseWithCancelledContextStillDeletes(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_release_cancelled" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + l := newLocker(t, db, table, lockerpostgres.Config{ + TTL: 30 * time.Second, Heartbeat: time.Hour, Owner: "owner-1", + AcquireBackoff: 10 * time.Millisecond, + AcquireDeadline: 300 * time.Millisecond, + }) + + ctx, cancel := context.WithCancel(context.Background()) + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice")) + cancel() + + l.ReleaseLocks(ctx, "anchor1") + + var count int + require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM "+table).Scan(&count)) + assert.Equal(t, 0, count, "a release on an already-cancelled context must still delete the leases") + require.NoError(t, l.AcquireLocks(context.Background(), "anchor2", "alice"), + "alice must be claimable again immediately, not only once the lease TTL expires") + l.ReleaseLocks(context.Background(), "anchor2") +} + func TestLocker_NilDB(t *testing.T) { _, err := lockerpostgres.New(nil, "t", lockerpostgres.Config{}, stubReplicaID{id: "owner"}) require.Error(t, err) @@ -378,3 +611,47 @@ func TestLocker_OwnerFromReplicaID(t *testing.T) { require.NoError(t, db.QueryRow("SELECT owner FROM "+table+" WHERE eid = $1", "alice").Scan(&owner)) assert.Equal(t, "replica-7", owner) } + +// TestLocker_NarrowingDeletesDroppedLeaseRows pins the row-level outcome the +// conformance suite checks through the interface. The acquisition statement only +// ever inserted, so an enrollment ID dropped from a live anchor kept its lease row. +// Both AssertLocksHeld and renewLeases count this replica's un-expired rows for the +// anchor and require exactly as many as the session recorded, so a single leftover +// row failed both: writes were rejected with "locks lost before write", the +// heartbeat gave up on its first tick, and once the TTL elapsed the abandoned lease +// expired and became claimable by another replica while this one still believed it +// held it. +func TestLocker_NarrowingDeletesDroppedLeaseRows(t *testing.T) { + db := startPostgres(t) + table := "test_eid_lease_narrowing" + cleanTable(t, db, table) + t.Cleanup(func() { cleanTable(t, db, table) }) + + cfg := lockerpostgres.Config{ + TTL: 30 * time.Second, Heartbeat: time.Hour, + AcquireBackoff: 10 * time.Millisecond, + AcquireDeadline: 300 * time.Millisecond, + Owner: "owner-1", + } + l := newLocker(t, db, table, cfg) + + ctx := context.Background() + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "alice", "bob")) + t.Cleanup(func() { l.ReleaseLocks(ctx, "anchor1") }) + require.NoError(t, l.AcquireLocks(ctx, "anchor1", "bob")) + + var eids []string + rows, err := db.Query("SELECT eid FROM "+table+" WHERE anchor = $1 AND owner = $2", "anchor1", "owner-1") + require.NoError(t, err) + defer func() { _ = rows.Close() }() + for rows.Next() { + var eid string + require.NoError(t, rows.Scan(&eid)) + eids = append(eids, eid) + } + require.NoError(t, rows.Err()) + assert.Equal(t, []string{"bob"}, eids, "the lease dropped from the anchor must be deleted, not left behind") + + // The counts the session, its heartbeat and its assertions all rely on now agree. + require.NoError(t, l.AssertLocksHeld(ctx, "anchor1")) +} diff --git a/token/services/storage/auditdb/store.go b/token/services/storage/auditdb/store.go index 316baca8da..bd17226037 100644 --- a/token/services/storage/auditdb/store.go +++ b/token/services/storage/auditdb/store.go @@ -338,9 +338,14 @@ func (d *StoreService) GetTokenRequests(ctx context.Context, txIDs []string) (ma // AcquireLocks acquires locks for the passed anchor and enrollment ids. // This can be used to prevent concurrent read/write access to the audit records of the passed enrollment ids. // The function respects context cancellation and deadlines, returning an error if the context is cancelled -// or times out before all locks can be acquired. This prevents indefinite blocking and enables fast failure -// in case of lock contention or deadlock scenarios. -// The implementation provides deadlock prevention through deterministic lock ordering (sorted by enrollment ID). +// or times out before all locks can be acquired, and the locker bounds the wait itself so a caller that +// passes no deadline still gets an answer. This prevents indefinite blocking and enables fast failure +// in case of lock contention. +// +// Deadlock freedom comes from the locker.Locker contract rather than from lock ordering alone: the +// enrollment IDs of one call are taken in a canonical order, and the same anchor may not later add IDs to +// the set it already holds, which is what would put held locks outside that order. See locker.Locker for +// the full contract; re-acquiring a live anchor with a wider set fails with ErrLockSetWidened. // Livelock prevention is handled by the caller through retry logic with exponential backoff. func (d *StoreService) AcquireLocks(ctx context.Context, anchor string, eIDs ...string) error { logger.DebugfContext(ctx, "Acquire locks for [%s:%v] enrollment ids", anchor, eIDs)