From 38c4383446e880154df0141c5b5c84fbcb9dce84 Mon Sep 17 00:00:00 2001 From: Dave Page Date: Tue, 11 Aug 2026 14:11:10 +0100 Subject: [PATCH 1/2] Back off when a batch fails for a reason no item can be charged for A failure belonging to a particular queue item is charged to it, and the queue's own machinery then holds it back: next_retry_at keeps it out of the claim and max_attempts eventually retires it. A failure belonging to the batch has none of that. Nothing is charged, deliberately, because billing a blameless item for a misconfigured provider would work through the queue retiring one innocent row per max_attempts cycles. The consequence was that those failures spun. The same rows were reclaimed and failed identically on the very next poll, indefinitely, filling the log at whatever rate worker_poll_interval allowed. An unset or misspelled pgedge_vectorizer.provider was enough to provoke it, as was a failed provider init or a dimension probe failure. Measured at the 200ms poll the new test uses, that is 100 failure cycles in a twenty second window. Nothing in the queue can help, since all of it is keyed on attempts moving and there is no item to move it against, so the wait itself is lengthened instead: five seconds after the first such failure, doubling to a five minute ceiling, cleared by a batch that gets through. That mirrors the extension-not-installed retry in the same function and the launcher's failed-start backoff, so it is a third instance of a shape already here rather than a new idea. queue_item_record_failure() now reports whether it found an item to charge, which is what distinguishes the two cases. A reload also clears it, because a reload is how a misconfigured provider gets corrected and waiting out a backoff the fix has already invalidated would be obtuse. That required handling a SIGHUP arriving during the wait before running any further work rather than at the top of the following iteration: processing first ran one more batch against the configuration just corrected, doubling the backoff again and logging the stale interval. The new test caught that, and asserts the reload resets to the floor. Closes #52 --- src/worker.c | 97 +++++++++++++++++-- test/t/005_batch_failure_backoff.pl | 139 ++++++++++++++++++++++++++++ 2 files changed, 230 insertions(+), 6 deletions(-) create mode 100644 test/t/005_batch_failure_backoff.pl diff --git a/src/worker.c b/src/worker.c index a021dc9..aa81b14 100644 --- a/src/worker.c +++ b/src/worker.c @@ -165,7 +165,7 @@ static void worker_sighup(SIGNAL_ARGS); static void queue_item_begin(int64 queue_id, int attempts, int max_attempts); static void queue_item_done(void); static void queue_item_note_error(void); -static void queue_item_record_failure(void); +static bool queue_item_record_failure(void); static void process_queue_batch(const char *dbname); static void cleanup_completed_items(const char *dbname); static void update_embedding(int64 chunk_id, const char *chunk_table, @@ -267,8 +267,13 @@ queue_item_done(void) * losing the bookkeeping is preferable to taking the worker down with it. The * item is cleared either way, so a persistent problem recording failures * cannot make the worker retry the same row forever on that account. + * + * Returns whether the failure belonged to a particular item. False means the + * fault was the batch's rather than any one row's, so nothing was charged and + * none of the queue's own machinery will hold the work back; the caller backs + * off instead. See the batch backoff in pgedge_vectorizer_worker_main(). */ -static void +static bool queue_item_record_failure(void) { int64 queue_id = failed_item_queue_id; @@ -278,7 +283,7 @@ queue_item_record_failure(void) char *quoted_reason; if (queue_id < 0) - return; + return false; exhausted = (failed_item_attempts + 1 >= failed_item_max_attempts); reason = (failed_item_error[0] != '\0') ? failed_item_error @@ -357,6 +362,14 @@ queue_item_record_failure(void) "queue item " INT64_FORMAT, queue_id); } PG_END_TRY(); + + /* + * The fault was this item's whether or not the record survived. A failure + * to write it is reported above and leaves the row to be tried again, + * which is the same position an uncharged attempt leaves it in, but it is + * not the batch-wide fault that the caller's backoff exists for. + */ + return true; } /* @@ -1082,6 +1095,28 @@ pgedge_vectorizer_worker_main(Datum main_arg) bool first_ext_check = true; #define EXT_RETRY_MAX 300000 /* Cap at 5 minutes */ + /* + * Batch failure backoff. + * + * A failure belonging to a particular queue item is charged to it, and the + * queue's own machinery then holds it back: next_retry_at keeps it out of + * the claim and max_attempts eventually retires it. A failure belonging to + * the batch rather than to any one row has none of that. Nothing is + * charged, deliberately, because billing a blameless item for a + * misconfigured provider would work through the queue retiring one + * innocent row per max_attempts cycles. So the same rows are reclaimed and + * fail identically on the very next poll, indefinitely, filling the log at + * whatever rate worker_poll_interval allows. An unset or misspelled + * pgedge_vectorizer.provider is enough to provoke it. + * + * Nothing in the queue can help here, since all of it is keyed on attempts + * moving and there is no item to move it against, so the wait itself is + * lengthened instead. Zero means no batch failure is outstanding. + */ + int batch_retry_interval = 0; +#define BATCH_RETRY_MIN 5000 /* First backoff: 5 seconds */ +#define BATCH_RETRY_MAX 300000 /* Cap at 5 minutes */ + /* Setup signal handlers */ pqsignal(SIGTERM, worker_sigterm); pqsignal(SIGHUP, worker_sighup); @@ -1131,6 +1166,14 @@ pgedge_vectorizer_worker_main(Datum main_arg) extension_exists = false; ext_retry_interval = 5000; + /* + * Retry the queue at once. A reload is how a misconfigured + * provider gets fixed, so making the operator wait out a backoff + * that their correction has already invalidated would be + * needlessly obtuse. + */ + batch_retry_interval = 0; + /* * Re-evaluate our quantum: a reload may have added databases or * lowered the cap, turning a resident worker into one that must @@ -1182,8 +1225,19 @@ pgedge_vectorizer_worker_main(Datum main_arg) /* Update process status */ pgstat_report_activity(STATE_IDLE, NULL); - /* Use longer wait time if extension not installed */ - wait_time = extension_exists ? pgedge_vectorizer_worker_poll_interval : ext_retry_interval; + /* + * Use a longer wait if the extension is not installed, or if the last + * batch failed for a reason no single item can be charged for. The + * backoff is floored at the poll interval so that a configuration + * with a long poll is never made to poll faster by failing. + */ + if (!extension_exists) + wait_time = ext_retry_interval; + else if (batch_retry_interval > 0) + wait_time = Max(batch_retry_interval, + pgedge_vectorizer_worker_poll_interval); + else + wait_time = pgedge_vectorizer_worker_poll_interval; /* Wait for work or timeout */ rc = WaitLatch(MyLatch, @@ -1197,6 +1251,18 @@ pgedge_vectorizer_worker_main(Datum main_arg) if (rc & WL_POSTMASTER_DEATH) proc_exit(1); + /* + * Apply a reload that arrived during the wait before doing any + * further work, by going back to the top where it is handled. + * Processing first would run one more batch against the + * configuration the operator has just corrected and charge them for + * its failure: the backoff would double again, and the interval + * logged would be the stale one rather than the fresh start the + * reload earns. + */ + if (got_sighup) + continue; + /* * Yield our slot once the service quantum expires, so that the * launcher can hand it to the next database in the rotation. A worker @@ -1233,6 +1299,12 @@ pgedge_vectorizer_worker_main(Datum main_arg) /* Perform automatic cleanup if enabled */ cleanup_completed_items(dbname); + + /* + * A batch that got through clears any backoff: whatever was wrong + * is no longer wrong, and there is no reason to keep waiting. + */ + batch_retry_interval = 0; } PG_CATCH(); { @@ -1259,8 +1331,21 @@ pgedge_vectorizer_worker_main(Datum main_arg) * its own and none can be started until this one is cleared. * Without it the attempt is discarded along with everything else * and the same row is reclaimed on the next poll, indefinitely. + * + * Nothing to charge means the fault was the batch's, so back off + * rather than spin: see the batch backoff declared at the top of + * this function. */ - queue_item_record_failure(); + if (!queue_item_record_failure()) + { + batch_retry_interval = (batch_retry_interval == 0) + ? BATCH_RETRY_MIN + : Min(batch_retry_interval * 2, BATCH_RETRY_MAX); + + elog(LOG, "pgedge_vectorizer worker for database \"%s\": batch " + "failed with nothing to charge it to, waiting %ds before " + "trying again", dbname, batch_retry_interval / 1000); + } /* Recheck extension status on error */ extension_exists = false; diff --git a/test/t/005_batch_failure_backoff.pl b/test/t/005_batch_failure_backoff.pl new file mode 100644 index 0000000..dc37f04 --- /dev/null +++ b/test/t/005_batch_failure_backoff.pl @@ -0,0 +1,139 @@ +# Copyright (c) 2025 - 2026, pgEdge, Inc. +# +# Verify that a batch which fails for a reason no single queue item can be +# charged for makes the worker back off, rather than reclaiming the same rows +# and failing identically on every poll. +# +# A failure belonging to a particular item is charged to it, and the queue then +# holds it back: next_retry_at keeps it out of the claim and max_attempts +# eventually retires it (see 004_queue_failure_accounting.pl). A failure +# belonging to the batch has none of that. Nothing is charged, deliberately, +# since billing a blameless item for a misconfigured provider would work +# through the queue retiring one innocent row per max_attempts cycles. So +# before this change the same rows were reclaimed on the very next poll and +# failed the same way, indefinitely: at the 200ms poll used here that is five +# failure cycles a second, for as long as the misconfiguration lasted. +# +# The fault injected is a provider that does not exist, which makes +# get_current_provider() raise. That is deliberate on two counts: it needs no +# network and no API key, and it is the most likely way for a real deployment +# to land here, since a single mistyped pgedge_vectorizer.provider does it. +# +# The queue row points at a chunk table that genuinely exists, so that the +# probe preceding the provider lookup succeeds. Were it missing, the failure +# would be charged to the item and the queue's own backoff would cover it, +# which is the case this test is specifically not about. + +use strict; +use warnings; + +# See the comment in 001_worker_coverage.pl about loading these at compile time. +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $dbname = 'batch_backoff'; + +my $node = PostgreSQL::Test::Cluster->new('vectorizer_batch_backoff'); +$node->init; +$node->append_conf( + 'postgresql.conf', qq( +shared_preload_libraries = 'pgedge_vectorizer' +pgedge_vectorizer.worker_poll_interval = 200 +pgedge_vectorizer.provider = 'no_such_provider' +max_worker_processes = 16 +)); + +$node->start; + +$node->safe_psql('postgres', "CREATE DATABASE $dbname"); +$node->safe_psql($dbname, 'CREATE EXTENSION vector'); +$node->safe_psql($dbname, 'CREATE EXTENSION pgedge_vectorizer'); + +# A real chunk table, so the dense-embedding probe succeeds and the batch gets +# as far as resolving the provider. +$node->safe_psql( + $dbname, q( +CREATE TABLE chunks ( + id BIGSERIAL PRIMARY KEY, + content TEXT, + token_count INT, + embedding vector(3), + sparse_embedding sparsevec(100) +); +INSERT INTO chunks (content, token_count) VALUES ('alpha beta gamma', 3); +)); + +# Name the database only once it is ready to be serviced, so that no worker can +# arrive before the extension exists and take its five second backoff instead. +$node->append_conf('postgresql.conf', + "pgedge_vectorizer.databases = '$dbname'\n"); +$node->reload; + +my $offset = (-s $node->logfile) // 0; + +$node->safe_psql( + $dbname, q( +INSERT INTO pgedge_vectorizer.queue (chunk_id, chunk_table, content, status) +VALUES (1, 'chunks', 'alpha beta gamma', 'pending') +)); + +# Wait for the first failure rather than assuming one has happened, so worker +# startup timing cannot decide the result. +my $deadline = time() + 30; +my $log = ''; + +while (time() < $deadline) +{ + $log = slurp_file($node->logfile, $offset); + + last if $log =~ /error in processing, continuing/; + + sleep 1; +} + +like($log, qr/error in processing, continuing/, + 'the batch does fail, so the rest of this test is measuring something'); + +# Twenty seconds covers the first three attempts of a 5s, 10s, 20s backoff. +# Unfixed, a 200ms poll manages of the order of a hundred in the same window. +sleep 20; +$log = slurp_file($node->logfile, $offset); + +my @cycles = ($log =~ /error in processing, continuing/g); + +cmp_ok(scalar(@cycles), '<=', 8, + 'a batch that cannot be charged to an item is retried a handful of times, not continuously'); + +cmp_ok(scalar(@cycles), '>=', 1, + 'the worker does keep retrying rather than giving up on the queue'); + +# The intervals themselves: the first backoff is the floor, and each failure +# thereafter doubles it. +my @waits = ($log =~ /waiting (\d+)s before trying again/g); + +cmp_ok(scalar(@waits), '>=', 2, + 'the backoff is reported so an operator can see why the queue has gone quiet'); + +is($waits[0], '5', 'the first backoff is the five second floor'); + +cmp_ok($waits[1], '>', $waits[0], + 'each successive failure waits longer than the last'); + +# A reload is how a misconfigured provider gets corrected, so it must not leave +# the operator waiting out a backoff their fix has already invalidated. +my $reload_offset = (-s $node->logfile) // 0; +$node->reload; +sleep 8; + +my $after_reload = slurp_file($node->logfile, $reload_offset); +my @after_waits = ($after_reload =~ /waiting (\d+)s before trying again/g); + +cmp_ok(scalar(@after_waits), '>=', 1, + 'the worker tries again promptly after a reload rather than waiting out the old backoff'); + +is($after_waits[0], '5', + 'a reload resets the backoff to the floor'); + +$node->stop; +done_testing(); From 64b5f8c46cf8ff42e809cc306ea7f058f39a4fd4 Mon Sep 17 00:00:00 2001 From: Mason Sharp Date: Tue, 11 Aug 2026 12:57:52 -0700 Subject: [PATCH 2/2] Log the wait the batch backoff will actually take The message printed the raw backoff, but the wait is floored at worker_poll_interval, so a long poll was understated in the one message meant to explain why the queue had gone quiet. Also note that skipping to the top of the loop on SIGHUP defers the service quantum check by one wait. --- src/worker.c | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/worker.c b/src/worker.c index aa81b14..4a598d5 100644 --- a/src/worker.c +++ b/src/worker.c @@ -1259,6 +1259,9 @@ pgedge_vectorizer_worker_main(Datum main_arg) * its failure: the backoff would double again, and the interval * logged would be the stale one rather than the fresh start the * reload earns. + * + * This defers the quantum check below by one wait, which is harmless: + * the quantum has no deadline, and a reload is what may have changed it. */ if (got_sighup) continue; @@ -1342,9 +1345,16 @@ pgedge_vectorizer_worker_main(Datum main_arg) ? BATCH_RETRY_MIN : Min(batch_retry_interval * 2, BATCH_RETRY_MAX); + /* + * Report the wait actually taken, floored at the poll interval + * just as the wait below is; the raw backoff would understate + * a long poll. + */ elog(LOG, "pgedge_vectorizer worker for database \"%s\": batch " "failed with nothing to charge it to, waiting %ds before " - "trying again", dbname, batch_retry_interval / 1000); + "trying again", dbname, + Max(batch_retry_interval, + pgedge_vectorizer_worker_poll_interval) / 1000); } /* Recheck extension status on error */