diff --git a/src/worker.c b/src/worker.c index a021dc9..4a598d5 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,21 @@ 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. + * + * 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; + /* * 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 +1302,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 +1334,28 @@ 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); + + /* + * 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, + Max(batch_retry_interval, + pgedge_vectorizer_worker_poll_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();