From 3632d77c9c585bee189b365d57b4e1e1e473e77b Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Tue, 18 Aug 2026 14:26:42 +1000 Subject: [PATCH] fix: retry the ochre vector connect until shutdown so RAG self-heals (#824) The daemon's connect to the platform pgvector appliance gave up after a fixed five attempts; once exhausted, startOchreVector returned and the ochre.vector.query subject stayed unregistered for the daemon's lifetime, so RAG never recovered after the appliance was torn down and re-adopted across a spinifex.target restart without a manual daemon bounce. Retry Ensure-then-Connect until the daemon context is cancelled instead, via a small generic retryUntilContext helper with a capped doubling backoff and throttled logging. Each attempt already re-resolves the live dial IP, so a re-adopted appliance on a new address is picked up and the query subject registers on its own. --- spinifex/daemon/ochre_deps.go | 77 +++++++++++++++++------------- spinifex/daemon/ochre_deps_test.go | 44 +++++++++++++++++ 2 files changed, 87 insertions(+), 34 deletions(-) diff --git a/spinifex/daemon/ochre_deps.go b/spinifex/daemon/ochre_deps.go index e19fbf543..0a6491abb 100644 --- a/spinifex/daemon/ochre_deps.go +++ b/spinifex/daemon/ochre_deps.go @@ -22,11 +22,13 @@ import ( // boot plus initdb on the smallest instance class. const ochreApplianceLaunchTimeout = 10 * time.Minute -// ochreStartupMaxAttempts bounds how many times startOchreVector retries the -// Ensure-then-Connect pair. A still-provisioning appliance or a host port -// not yet routed is transient, but the retry still needs a ceiling so a -// genuine misconfiguration eventually gives up instead of retrying forever. -const ochreStartupMaxAttempts = 5 +// ochreStartupLogAttempts and ochreStartupLogEvery throttle the connect-retry +// log: warn on the first few attempts, then only every Nth, so a persistently +// unreachable appliance leaves a periodic breadcrumb without flooding the log. +const ( + ochreStartupLogAttempts = 3 + ochreStartupLogEvery = 20 +) // ochreStartupInitialBackoff and ochreStartupMaxBackoff bound the wait // between attempts: it doubles from the initial value and is capped so a @@ -104,8 +106,9 @@ func (d *Daemon) startOchreVector() { backend, err := d.connectOchreAppliance(appliance) if err != nil { - slog.Warn("Ochre vector store disabled: platform appliance not reachable after retrying", - "attempts", ochreStartupMaxAttempts, "err", err) + // connectOchreAppliance only returns on daemon shutdown now; there is + // no give-up path, so an error here is teardown, not a failure to wire. + slog.Info("Ochre vector store: appliance connect abandoned on shutdown", "err", err) return } @@ -217,43 +220,49 @@ func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, _ *handlers_o return &handlers_ochrevector.TeardownApplianceResponse{}, nil } -// connectOchreAppliance drives Ensure then Connect as one retryable unit, up -// to ochreStartupMaxAttempts times with exponential backoff between -// attempts. Previously a single provisioning timeout -- the appliance still -// booting, or its host port not yet routed -- hard-disabled the feature for -// the daemon's entire lifetime; this bounds that to a handful of attempts -// instead of exactly one, while still eventually giving up on a persistent -// failure rather than retrying forever. +// connectOchreAppliance drives Ensure-then-Connect until it succeeds or the +// daemon shuts down, so a torn-down-then-readopted appliance heals on its own +// rather than disabling the feature permanently after a few early failures. func (d *Daemon) connectOchreAppliance(appliance *handlers_ochrevector.Appliance) (handlers_ochrevector.VectorBackend, error) { - backoff := ochreStartupInitialBackoff - var lastErr error - for attempt := 1; attempt <= ochreStartupMaxAttempts; attempt++ { - if d.ctx.Err() != nil { - return nil, d.ctx.Err() - } + return retryUntilContext(d.ctx, ochreStartupInitialBackoff, ochreStartupMaxBackoff, + func(attempt int, backoff time.Duration, err error) { + if attempt <= ochreStartupLogAttempts || attempt%ochreStartupLogEvery == 0 { + slog.Warn("Ochre vector store: appliance not reachable, will keep retrying", + "attempt", attempt, "backoff", backoff, "err", err) + } + }, + func() (handlers_ochrevector.VectorBackend, error) { + return d.ensureAndConnectOchreApplianceOnce(appliance) + }) +} - backend, err := d.ensureAndConnectOchreApplianceOnce(appliance) +// retryUntilContext calls attempt until it succeeds or ctx is cancelled, +// sleeping a doubling backoff (capped at maxBackoff) between failures and +// reporting each to log. No attempt ceiling: ctx cancellation is the only exit. +func retryUntilContext[T any](ctx context.Context, initialBackoff, maxBackoff time.Duration, + log func(attempt int, backoff time.Duration, err error), attempt func() (T, error)) (T, error) { + backoff := initialBackoff + for n := 1; ; n++ { + if err := ctx.Err(); err != nil { + var zero T + return zero, err + } + v, err := attempt() if err == nil { - return backend, nil + return v, nil } - lastErr = err - - if attempt == ochreStartupMaxAttempts { - break - } - slog.Warn("Ochre vector store: appliance not ready, retrying", - "attempt", attempt, "max_attempts", ochreStartupMaxAttempts, "backoff", backoff, "err", lastErr) + log(n, backoff, err) select { - case <-d.ctx.Done(): - return nil, d.ctx.Err() + case <-ctx.Done(): + var zero T + return zero, ctx.Err() case <-time.After(backoff): } backoff *= 2 - if backoff > ochreStartupMaxBackoff { - backoff = ochreStartupMaxBackoff + if backoff > maxBackoff { + backoff = maxBackoff } } - return nil, lastErr } // ensureAndConnectOchreApplianceOnce runs a single Ensure-then-Connect diff --git a/spinifex/daemon/ochre_deps_test.go b/spinifex/daemon/ochre_deps_test.go index 8a7c566b2..fcb9d5859 100644 --- a/spinifex/daemon/ochre_deps_test.go +++ b/spinifex/daemon/ochre_deps_test.go @@ -6,7 +6,9 @@ package daemon import ( "context" + "errors" "testing" + "time" "github.com/mulgadc/spinifex/spinifex/config" handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector" @@ -33,6 +35,48 @@ func TestStartOchreVector_DisabledSkipsConstruction(t *testing.T) { assert.Nil(t, d.natsSubscriptions, "disabled path must not touch the subscriptions map") } +// TestRetryUntilContext_RetriesPastOldCeilingThenSucceeds proves the connect +// loop no longer gives up after the old fixed ceiling: it keeps retrying until +// the appliance is reachable, so RAG heals once a re-adopted appliance returns. +func TestRetryUntilContext_RetriesPastOldCeilingThenSucceeds(t *testing.T) { + const failuresBeforeSuccess = 7 // deliberately > the old 5-attempt ceiling + + calls, logged := 0, 0 + got, err := retryUntilContext(context.Background(), time.Millisecond, 2*time.Millisecond, + func(attempt int, backoff time.Duration, err error) { logged++ }, + func() (int, error) { + calls++ + if calls <= failuresBeforeSuccess { + return 0, errors.New("appliance not reachable") + } + return 42, nil + }) + + require.NoError(t, err) + assert.Equal(t, 42, got) + assert.Equal(t, failuresBeforeSuccess+1, calls, "must retry past the old fixed ceiling") + assert.Equal(t, failuresBeforeSuccess, logged, "each failure reported to log once") +} + +// TestRetryUntilContext_StopsOnContextCancel proves daemon shutdown is the +// loop's exit: a cancel while it is backing off returns context.Canceled +// promptly rather than sleeping out the backoff or starting another attempt. +func TestRetryUntilContext_StopsOnContextCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + calls := 0 + got, err := retryUntilContext(ctx, time.Hour, time.Hour, + func(attempt int, backoff time.Duration, err error) { cancel() }, + func() (int, error) { + calls++ + return 0, errors.New("still down") + }) + + require.ErrorIs(t, err, context.Canceled) + assert.Zero(t, got) + assert.Equal(t, 1, calls, "cancel during backoff stops before the next attempt") +} + // TestHandleOchreApplianceTeardown_NilApplianceRefuses proves the handler // refuses with a clear error rather than a nil-pointer panic when the // appliance never came up (disabled, still starting, or already torn down by