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

### 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:

```sql
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:
```cs
Task 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:
```sql
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.

Start by tracing CollectorProcessor and its DeleteExpiresAsync calls, then read IDataStorage's AcquireLockAsync, RenewLockAsync, and ReleaseLockAsync implementations alongside the existing UseStorageLock retry path. The maintainer must first choose between a collector storage lease and provider-specific SKIP LOCKED claiming; done means the agreed behavior is opt-in, coordinates replicas, handles lease expiry, and covers the supported storage providers.

Written by the indexing model from the issue text.