Coordinate expired-message cleanup across replicas using storage locks or SKIP LOCKED

[3ldar]

Question Summary

Coordinate expired-message cleanup across replicas using storage locks or SKIP LOCKED

Context

In Kubernetes deployments, multiple CAP application replicas can run
CollectorProcessor against the same storage schema.

In CAP 10.0.1, every instance periodically processes both the published and
received tables. Each collector repeatedly calls DeleteExpiresAsync in
batches of 1,000 until no rows remain.

The PostgreSQL implementation currently uses:

DELETE FROM {table}
WHERE "Id" IN (
    SELECT "Id"
    FROM {table}
    WHERE "ExpiresAt" < @timeout
      AND "StatusName" IN ('Succeeded', 'Failed')
    LIMIT @batchCount
)

There is no coordination or row locking around candidate selection. When
several replicas clean the same schema concurrently, they can select overlapping
IDs. This creates avoidable lock waits, redundant DELETE attempts, database
traffic, and noisy retries/logging.

Relevant existing mechanism

CAP already exposes the following storage-lock operations through
IDataStorage:

  Task<bool> AcquireLockAsync(...);
  Task RenewLockAsync(...);
  Task ReleaseLockAsync(...);

UseStorageLock currently applies to failed-message retry processing. I propose
reusing the same storage-lock abstraction for collector cleanup.

Preferred solution: collector storage lease

Add an opt-in collector lock, either through a dedicated option such as:

UseStorageLockForCollector

or by explicitly extending and documenting UseStorageLock.

Suggested behavior:

1. Acquire a collector-specific lock before processing published and
   received.

2. Use a key scoped to the CAP storage/schema.

3. If another instance owns the lock, skip the current collection cycle rather
   than retrying in a tight loop.

4. Renew the lease while cleanup is running because a large backlog may take
   longer than the initial TTL.

5. Release the lease in finally.

6. Allow another instance to recover after TTL expiry if the owning pod dies.

A dedicated option may be safer because changing the meaning of
UseStorageLock could affect existing deployments.

Alternative: provider-specific row claiming

For PostgreSQL and other databases supporting SKIP LOCKED, candidates could
instead be claimed atomically:

  WITH candidates AS (
      SELECT "Id"
      FROM {table}
      WHERE "ExpiresAt" < @timeout
        AND "StatusName" IN ('Succeeded', 'Failed')
      FOR UPDATE SKIP LOCKED
      LIMIT @batchCount
  )
  DELETE FROM {table} AS target
  USING candidates
  WHERE target."Id" = candidates."Id";

This allows multiple collectors to work concurrently while ensuring that each
batch processes different rows.

The SQL would need provider-specific implementations. For databases without
equivalent row-locking support, the storage-lock approach provides a portable
fallback.

Trade-offs

Storage lock

• Lowest database noise.
• Portable through the existing CAP storage abstraction.
• Only one collector processes a storage schema at a time.
• Requires lease renewal and crash recovery.

SKIP LOCKED

• Allows parallel cleanup of large backlogs.
• Prevents collectors from waiting on the same candidate rows.
• Requires provider-specific SQL.

For routine periodic cleanup, the storage lock appears to be the safer default.
SKIP LOCKED may be useful when parallel backlog cleanup is desirable.

• The feature is backward-compatible and opt-in unless maintainers prefer to
  change the existing default.

I would be willing to contribute a PR after agreeing on the preferred
configuration and cross-provider behavior.

[yang-xiaodong]

Do you want to make a PR using FOR UPDATE SKIP LOCKED for postgres?