feat: replace Redis queue with Postgres-native atomic job ownership - #582
feat: replace Redis queue with Postgres-native atomic job ownership#582acardace wants to merge 16 commits into
Conversation
f5ffbc3 to
8fb15ea
Compare
| @@ -0,0 +1,3 @@ | |||
| TODO: | |||
There was a problem hiding this comment.
Is this to be committed, or just a planning doc?
|
I agree that this PR is trying to solve a real correctness problem, but I have a concern about the architectural direction it takes. This design collapses queue ownership directly into the Postgres batch store ( I do not think this is the direction we want. One of the goals of this project is to keep storage roles independently replaceable. We have already been moving in the direction of making Redis optional by providing alternative backend implementations, not by collapsing multiple roles into a single storage model. If we merge this design, we effectively give up queue-level pluggability and make SQL-backed ownership semantics the only realistic model. The better direction, in my view, is to separate delivery from ownership:
That would preserve the correctness properties this PR is aiming for without destroying the queue abstraction. |
As a follow-up, maybe it's better to focus instead on decoupling job level ownership (low TPS) and transport (higher TPS) into two distinct interfaces, if we break up Ownership can stay in an RDBMS (we already require one) and then transport can stay pluggable. I think I agree with all the other points made |
Yes, I agree. This makes the direction much clearer. |
|
I agree with @j-mok-dev and @wseaton - added a comment in the design issue #572 (comment) |
|
This PR is marked as stale after 21d of inactivity. After an additional 14d of inactivity (7d to become rotten, then 7d more), it will be closed. To prevent this PR from being closed, add a comment or remove the |
|
Comment to remove stale label. |
The ExtraColumns scan destinations and return maps used map[string]string, forcing all extra columns to be strings. Change to map[string]any so that numeric types (e.g. BIGINT) are scanned as their native Go types without string conversion. Signed-off-by: Antonio Cardace <acardace@redhat.com>
Add processor_id (TEXT) and priority (BIGINT) columns to the batch_items schema. processor_id records which processor pod owns an in-progress job. priority stores SLO.UnixMicro() for queue ordering (lower = earlier deadline = higher priority). Add partial indexes for efficient dequeue (unclaimed validating jobs ordered by priority) and processor ownership lookup (crash recovery by processor_id). Add corresponding fields to BatchItem and wire them through the Postgres batch DB client as extra columns. Signed-off-by: Antonio Cardace <acardace@redhat.com>
…table Implement the priority queue interface using the batch_items table as a native Postgres queue. Jobs with status 'validating' and processor_id IS NULL are the queue entries. PQDequeue uses a CTE with SELECT FOR UPDATE SKIP LOCKED + UPDATE in a single atomic statement to claim jobs — no transaction needed. PQEnqueue sends a NOTIFY to wake listening processors. PQDelete atomically transitions unclaimed jobs to cancelled using FOR UPDATE SKIP LOCKED to prevent races with concurrent dequeue. Signed-off-by: Antonio Cardace <acardace@redhat.com>
When db_client.type is postgresql, create a PostgresBatchQueueClient that replaces the Redis priority queue. The processor's hostname (os.Hostname) is used as the processor ID for job claiming. For non-processor components (apiserver, GC), the processor ID is empty since they only enqueue/delete, never dequeue. Signed-off-by: Antonio Cardace <acardace@redhat.com>
Add recoverOwnedJobs which queries the DB for non-terminal jobs owned by this processor (via processor_id column) and recovers them at startup. This handles both container-level crashes and pod-level restarts within a StatefulSet. Replace the filesystem-based recoverStaleJobs with recoverOwnedJobs in the processor Run() flow. The DB is now the source of truth for crash recovery, not the local workdir. Add ProcessorID filter to BatchQuery so DBGet can filter by processor_id. Signed-off-by: Antonio Cardace <acardace@redhat.com>
Add a pod watcher that uses client-go informers to watch processor pods via label selectors. On pod add/delete events, it computes the set of live processor pod names and calls a handler function. This enables the reconciler to detect processor crashes immediately via Kubernetes events instead of periodic polling with staleness thresholds. Signed-off-by: Antonio Cardace <acardace@redhat.com>
BatchToDBItem now derives Priority (SLO as UnixMicro) from the batch's CompletionWindow and CreatedAt fields. This ensures the priority column is set when the apiserver stores new batch jobs, enabling the Postgres queue to order jobs by SLO deadline. Signed-off-by: Antonio Cardace <acardace@redhat.com>
Replace the processor Deployment with a StatefulSet to provide stable pod identities (processor-0, processor-1, ...) required for DB-based crash recovery. Each processor uses its hostname (stable across restarts) as the processor_id when claiming jobs. Add a headless Service as required by the StatefulSet API. Use podManagementPolicy: Parallel so replicas scale up/down concurrently instead of sequentially. Signed-off-by: Antonio Cardace <acardace@redhat.com>
Replace all usages of the TagSLO tag (slo_unix_micro) with the Priority field on BatchItem. The converter now derives Priority from CompletionWindow and CreatedAt. The reconciler and recovery code read Priority directly instead of parsing from tags. Remove the TagSLO constant and the SLO tag from job creation. Signed-off-by: Antonio Cardace <acardace@redhat.com>
The processor_id column in batch_items and the Kubernetes pod watcher replace the Redis in-flight hash and periodic heartbeat for job ownership tracking. Remove: - InFlightClient interface and InFlightEntry struct - Redis InFlight implementation (redis_inflight.go) - Mock InFlight client (mock_inflight_client.go) - InFlight field from Clientset and Processor - heartbeat goroutine and deleteInFlight function - HeartbeatInterval config field and its YAML/helm references - All InFlight-related tests Signed-off-by: Antonio Cardace <acardace@redhat.com>
Postgres is now the only supported DB backend. Remove the Redis CRUD implementation for batch and file items, update default DB type from redis to postgresql, and remove tests for Redis/Valkey DB configurations. The Redis exchange client (events, status, priority queue) is kept for backward compatibility. Signed-off-by: Antonio Cardace <acardace@redhat.com>
startProcessorObsPortForward reads scanDone when kubectl exits with an error, then the defer cleanup reads it again, blocking forever on the empty buffered channel. Fix by setting scanDone = nil after draining it so the defer skips the second read. Also wait for the processor pod to be Ready (kubectl wait) before attempting the port-forward, avoiding the pod-not-running race during pod transitions. Signed-off-by: Antonio Cardace <acardace@redhat.com>
The RollingRestartOrphan test sets inter-token-latency to 30s to keep requests in-flight during SIGTERM. After the restart, the processor self-recovers and re-processes the job — but the sim was still slow, causing the test to take unnecessarily long. Restore sim latency to 100ms after the rollout completes so re-processed requests complete quickly. Signed-off-by: Antonio Cardace <acardace@redhat.com>
PQDelete now atomically transitions queued jobs to cancelled (with cancelled_at) in a single SQL statement. When PQDelete succeeds (nDeleted > 0), the handler re-reads the updated batch and returns it directly — no separate DBUpdate needed. For in-progress jobs (nDeleted == 0), the handler uses CAS on DBUpdate to set cancelling status, preventing a race where the processor could complete the job between the handler's read and write. Signed-off-by: Antonio Cardace <acardace@redhat.com>
…ueueing The reconciler re-enqueued all orphans with a valid SLO, including cancelling jobs. This lost the user's cancel intent. Add status-aware triage: cancelling orphans transition to cancelled. Refactor expireOrphan into transitionOrphan to avoid duplication. Update e2e OrphanRecovery tests for the new architecture: - HardCrashInProgress: expects re-enqueue then completion - CancellingOrphan: expects cancelled Signed-off-by: Antonio Cardace <acardace@redhat.com>
Add an epoch column to batch_items that acts as a fencing token, incremented on every ownership change. This prevents a zombie processor (stuck in Terminating, unfrozen after a network partition) from overwriting state written by the new owner. Schema: - Add epoch BIGINT NOT NULL DEFAULT 0 to batch_items Epoch is incremented on: - PQDequeue: new processor claims a job - PQEnqueue: GC re-enqueues an orphan back to the queue - PQDelete: cancel from queue - transitionOrphan: GC transitions an orphan to a terminal status (BumpEpoch flag on BatchItem triggers epoch = epoch + 1 in SET) Epoch is checked (not incremented) on: - All processor writes via StatusUpdater -> DBUpdate: WHERE epoch = N ensures a zombie whose job was reclaimed cannot overwrite the new owner's state. Sequential writes by the same processor reuse the same epoch (like a Raft term). Implementation: - BatchItem.Epoch carries the fencing token through the write path - BatchItem.BumpEpoch signals DBUpdate to atomically increment epoch - pgCore.update() accepts extraConditions (WHERE) and rawSetClauses - MockDBClient.DBUpdate checks both expectedStatus (CAS) and Epoch - TestEpochFencing covers matching, stale, and GC-bump scenarios Signed-off-by: Antonio Cardace <acardace@redhat.com>
8fb15ea to
7758208
Compare
4 participants
Summary
Replaces the Redis-based priority queue and in-flight tracking with a Postgres-native queue using
SELECT ... FOR UPDATE SKIP LOCKED. Job dequeue and ownership become a single atomic transaction — no gap, no distributed bookkeeping.Design: #572
Changes
Postgres-native queue
processor_id,priority, andepochcolumns tobatch_itemsPQDequeue: atomic CTE that claims jobs withFOR UPDATE SKIP LOCKEDand setsprocessor_idPQEnqueue: re-enqueue with guards against resurrecting terminal or already-queued jobsPQDelete: atomic cancel-from-queue withcancelled_attimestamppriority ASC WHERE processor_id IS NULL) and recovery (processor_id WHERE NOT NULL)Processor: Deployment → StatefulSet
processor-0,processor-1, ...) viaos.Hostname()Job recovery
recoverOwnedJobs: on startup, processor queries DB for non-terminal jobs byprocessor_idand recovers them — no GC dependency for same-pod restartsrecoverStaleJobs(filesystem scan) — superseded by DB-based recoveryGC: event-driven orphan recovery
cancellingorphans tocancelledHasProcessorIDfilter — reconciler only fetches owned jobs, not the entire queueEpoch-based fencing
WHERE epoch = N— a zombie whose job was reclaimed cannot overwrite the new owner's stateBumpEpochflagCancel handler
PQDeletesetscancelled+cancelled_atatomically; handler re-reads and returns current stateexpectedStatus) to prevent overwriting terminal statusComplexity reduction
BatchDBClient,FileDBClient,InFlightClient, heartbeat goroutineTesting