Delivery to external warehouses (#8107) tracks progress purely through S3 prefixes: pending objects live under events/, delivered ones move to archive/, and rejected ones to failed/. Two problems follow from having no per-object state.
Transient failures cause permanent data loss
An object is moved to failed/ on its first rejection, with no retry. A rejection code like 41 or 117 usually means bad content, but the same codes surface for reasons that would succeed on a second attempt. failed/ is covered by the bucket's expire-objects lifecycle rule (prefix "", 30 days), so those events are deleted permanently with no operator action available.
Concurrent runs are possible
deliver_events_for_connection is bounded by a time budget so it finishes inside its 9-minute task timeout, which is what currently prevents overlap. If a run does exceed the timeout, the task processor abandons the still-running thread (future.result(timeout=...) then shuts the executor down with wait=False) and retries the task, so two runs can deliver the same objects — duplicate inserts, and a race between copy_object/delete_object where the loser gets a 404.
Proposal
Add a per-object delivery record, keyed on (connection, s3_key), holding delivered_at, attempts and last_error. That gives:
- Retries with a cap. Transient failures are retried; an object is only abandoned after N attempts, so a blip no longer costs the data.
- Claim/lease as mutual exclusion.
UPDATE ... SET claimed_at = now() WHERE s3_key = ... AND claimed_at IS NULL is atomic, so concurrent runs cannot take the same object — finer-grained than a per-connection lock and with no new Redis keys.
- Bounded batches. "Claim up to N" replaces the time budget.
- Audit. What was delivered, when, and why anything failed, without listing S3.
Delivery stays at-least-once: the record is written after a successful insert, so a crash in between replays the object.
Delivery to external warehouses (#8107) tracks progress purely through S3 prefixes: pending objects live under
events/, delivered ones move toarchive/, and rejected ones tofailed/. Two problems follow from having no per-object state.Transient failures cause permanent data loss
An object is moved to
failed/on its first rejection, with no retry. A rejection code like 41 or 117 usually means bad content, but the same codes surface for reasons that would succeed on a second attempt.failed/is covered by the bucket'sexpire-objectslifecycle rule (prefix"", 30 days), so those events are deleted permanently with no operator action available.Concurrent runs are possible
deliver_events_for_connectionis bounded by a time budget so it finishes inside its 9-minute task timeout, which is what currently prevents overlap. If a run does exceed the timeout, the task processor abandons the still-running thread (future.result(timeout=...)then shuts the executor down withwait=False) and retries the task, so two runs can deliver the same objects — duplicate inserts, and a race betweencopy_object/delete_objectwhere the loser gets a 404.Proposal
Add a per-object delivery record, keyed on
(connection, s3_key), holdingdelivered_at,attemptsandlast_error. That gives:UPDATE ... SET claimed_at = now() WHERE s3_key = ... AND claimed_at IS NULLis atomic, so concurrent runs cannot take the same object — finer-grained than a per-connection lock and with no new Redis keys.Delivery stays at-least-once: the record is written after a successful insert, so a crash in between replays the object.