From f733fed177dc4f5d4c1cb0c1a6323d2474df46e9 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Tue, 25 Aug 2026 11:18:13 +1000 Subject: [PATCH] fix: preserve ochre teardown metadata for auto-reconcile (#863) Teardown no longer purges the index registry or KB/DataSource records by default, mirroring how they already survive a restart; a --purge-metadata flag restores the old full wipe for an intentional destroy. On the next appliance connect, a new reconcile pass re-creates any registry index whose pg table is missing and re-enqueues ingest from its persisted SourceSpecs, so a torn-down appliance heals without operator action and keeps its KB id. --- cmd/spinifex/cmd/admin_ochre_appliance.go | 28 +- .../cmd/admin_ochre_appliance_test.go | 69 ++++- spinifex/daemon/ochre_deps.go | 57 +++- spinifex/daemon/ochre_deps_test.go | 261 ++++++++++++++++++ spinifex/handlers/ochrevector/appliance.go | 70 +++-- .../handlers/ochrevector/appliance_test.go | 138 +++++++-- spinifex/handlers/ochrevector/backend.go | 6 + .../handlers/ochrevector/fake_backend_test.go | 7 + spinifex/handlers/ochrevector/ingest.go | 9 + spinifex/handlers/ochrevector/ingest_test.go | 28 ++ spinifex/handlers/ochrevector/pgx_backend.go | 19 ++ .../handlers/ochrevector/pgx_backend_test.go | 12 + 12 files changed, 632 insertions(+), 72 deletions(-) diff --git a/cmd/spinifex/cmd/admin_ochre_appliance.go b/cmd/spinifex/cmd/admin_ochre_appliance.go index 302997dda..342ddf0e1 100644 --- a/cmd/spinifex/cmd/admin_ochre_appliance.go +++ b/cmd/spinifex/cmd/admin_ochre_appliance.go @@ -29,10 +29,13 @@ var ochreApplianceTeardownCmd = &cobra.Command{ instance, this node's VPC host port, and its KV record. This is a destructive, cluster-wide operator action -- every account's vector -index data stored in the appliance is lost with it, and it is not -automatically rebuilt. A future daemon startup re-provisions a fresh -appliance (a new instance, a new password) only if Ochre remains enabled; -existing index registry records are not restored. +index data stored in the appliance is lost with it. By default the index +registry and KB/DataSource metadata survive, exactly as they survive a +restart: a future daemon startup re-provisions a fresh appliance (a new +instance, a new password) and reconciles each surviving index by re-creating +it and re-ingesting from its recorded source, with no operator action and no +change to its id. --purge-metadata additionally wipes that metadata for an +intentional full destroy, after which nothing is auto-restored. Requires --confirm.`, Run: runOchreApplianceTeardown, @@ -44,40 +47,47 @@ func init() { ochreApplianceTeardownCmd.Flags().Bool("confirm", false, "Required: confirms you intend to destroy the shared platform appliance and every account's vector index data with it") + ochreApplianceTeardownCmd.Flags().Bool("purge-metadata", false, + "Also wipe the index registry and KB/DataSource records, disabling auto-restore on the next re-provision") } // applianceTeardownFn indirects the NATS call so runApplianceTeardownGuarded // is testable without a live daemon, mirroring vectorServiceFn/ // endpointServiceFn. -var applianceTeardownFn = func(ctx context.Context) error { +var applianceTeardownFn = func(ctx context.Context, purgeMetadata bool) error { _, nc, err := loadConfigAndConnectFn() if err != nil { return err } defer nc.Close() _, err = utils.NATSRequest[handlers_ochrevector.TeardownApplianceResponse](ctx, nc, - handlers_ochrevector.SubjectTeardownAppliance, &handlers_ochrevector.TeardownApplianceRequest{}, applianceTeardownTimeout, utils.GlobalAccountID) + handlers_ochrevector.SubjectTeardownAppliance, &handlers_ochrevector.TeardownApplianceRequest{PurgeMetadata: purgeMetadata}, + applianceTeardownTimeout, utils.GlobalAccountID) return err } // runApplianceTeardownGuarded is the testable core of 'ochre appliance // teardown': it refuses without --confirm before making any NATS call, so a // missing flag never even opens a connection to the cluster. -func runApplianceTeardownGuarded(ctx context.Context, confirmed bool, teardown func(context.Context) error) (string, error) { +func runApplianceTeardownGuarded(ctx context.Context, confirmed, purgeMetadata bool, teardown func(context.Context, bool) error) (string, error) { if !confirmed { return "", errors.New("refusing to tear down the shared platform appliance without --confirm: " + "this destroys every account's vector index data and is not automatically rebuilt") } - if err := teardown(ctx); err != nil { + if err := teardown(ctx, purgeMetadata); err != nil { return "", err } + if purgeMetadata { + return "✅ Platform appliance torn down, index/KB metadata purged.", nil + } return "✅ Platform appliance torn down.", nil } func runOchreApplianceTeardown(cmd *cobra.Command, _ []string) { confirmed, _ := cmd.Flags().GetBool("confirm") + purgeMetadata, _ := cmd.Flags().GetBool("purge-metadata") - msg, err := runApplianceTeardownGuarded(context.Background(), confirmed, applianceTeardownFn) + msg, err := runApplianceTeardownGuarded(context.Background(), confirmed, purgeMetadata, applianceTeardownFn) if err != nil { fmt.Fprintf(os.Stderr, "%v\n", err) ochreExit(1) diff --git a/cmd/spinifex/cmd/admin_ochre_appliance_test.go b/cmd/spinifex/cmd/admin_ochre_appliance_test.go index 6eaa9ec0b..a0e85930d 100644 --- a/cmd/spinifex/cmd/admin_ochre_appliance_test.go +++ b/cmd/spinifex/cmd/admin_ochre_appliance_test.go @@ -9,6 +9,7 @@ import ( "errors" "testing" + "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/stretchr/testify/require" ) @@ -17,12 +18,12 @@ import ( // reach a NATS call, let alone destroy the appliance. func TestRunApplianceTeardownGuarded_RefusesWithoutConfirm(t *testing.T) { called := false - teardown := func(context.Context) error { + teardown := func(context.Context, bool) error { called = true return nil } - _, err := runApplianceTeardownGuarded(context.Background(), false, teardown) + _, err := runApplianceTeardownGuarded(context.Background(), false, false, teardown) require.Error(t, err) require.ErrorContains(t, err, "--confirm") require.False(t, called, "teardown must not run without --confirm") @@ -32,12 +33,12 @@ func TestRunApplianceTeardownGuarded_RefusesWithoutConfirm(t *testing.T) { // drives the teardown call through and reports success. func TestRunApplianceTeardownGuarded_ConfirmedInvokesTeardown(t *testing.T) { called := false - teardown := func(context.Context) error { + teardown := func(context.Context, bool) error { called = true return nil } - msg, err := runApplianceTeardownGuarded(context.Background(), true, teardown) + msg, err := runApplianceTeardownGuarded(context.Background(), true, false, teardown) require.NoError(t, err) require.True(t, called) require.Contains(t, msg, "torn down") @@ -47,19 +48,49 @@ func TestRunApplianceTeardownGuarded_ConfirmedInvokesTeardown(t *testing.T) { // failure (e.g. the daemon reporting no appliance up) surfaces to the caller // rather than being swallowed behind a generic success message. func TestRunApplianceTeardownGuarded_TeardownErrorSurfaces(t *testing.T) { - teardown := func(context.Context) error { + teardown := func(context.Context, bool) error { return errors.New("ochrevector: platform appliance is not enabled or not up on this node") } - _, err := runApplianceTeardownGuarded(context.Background(), true, teardown) + _, err := runApplianceTeardownGuarded(context.Background(), true, false, teardown) require.Error(t, err) require.ErrorContains(t, err, "not enabled or not up") } +// TestRunApplianceTeardownGuarded_PurgeMetadataThreadsThrough proves the +// --purge-metadata flag reaches the teardown call and is reflected in the +// success message, so an operator can tell which mode ran. +func TestRunApplianceTeardownGuarded_PurgeMetadataThreadsThrough(t *testing.T) { + var gotPurge bool + teardown := func(_ context.Context, purgeMetadata bool) error { + gotPurge = purgeMetadata + return nil + } + + msg, err := runApplianceTeardownGuarded(context.Background(), true, true, teardown) + require.NoError(t, err) + require.True(t, gotPurge, "purgeMetadata must reach the teardown call") + require.Contains(t, msg, "metadata purged") +} + +// TestApplianceTeardownFn_ThreadsPurgeMetadataOverNATS drives the real +// (unstubbed) applianceTeardownFn through runApplianceTeardownGuarded, so the +// actual NATSRequest call -- including the PurgeMetadata field on the wire +// request -- executes rather than a test double standing in for it. No +// daemon is subscribed to the teardown subject in this test, so the call +// fails fast; the point is that it's reached and built with PurgeMetadata set. +func TestApplianceTeardownFn_ThreadsPurgeMetadataOverNATS(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + stubConnect(t, fakeClusterConfig(), nc, nil) + + _, err := runApplianceTeardownGuarded(context.Background(), true, true, applianceTeardownFn) + require.Error(t, err, "no daemon is subscribed to the teardown subject in this test") +} + // withApplianceTeardown swaps applianceTeardownFn for one that never dials // NATS, so the Run wrapper's real flag/exit control flow is exercised // without a live daemon. -func withApplianceTeardown(t *testing.T, fn func(context.Context) error) { +func withApplianceTeardown(t *testing.T, fn func(context.Context, bool) error) { t.Helper() orig := applianceTeardownFn t.Cleanup(func() { applianceTeardownFn = orig }) @@ -67,7 +98,7 @@ func withApplianceTeardown(t *testing.T, fn func(context.Context) error) { } func TestRunOchreApplianceTeardown_MissingConfirmExits1(t *testing.T) { - withApplianceTeardown(t, func(context.Context) error { + withApplianceTeardown(t, func(context.Context, bool) error { t.Fatal("teardown must not be called without --confirm") return nil }) @@ -78,7 +109,7 @@ func TestRunOchreApplianceTeardown_MissingConfirmExits1(t *testing.T) { } func TestRunOchreApplianceTeardown_ConfirmedPrintsSuccess(t *testing.T) { - withApplianceTeardown(t, func(context.Context) error { return nil }) + withApplianceTeardown(t, func(context.Context, bool) error { return nil }) cmd := *ochreApplianceTeardownCmd require.NoError(t, cmd.Flags().Set("confirm", "true")) @@ -92,7 +123,7 @@ func TestRunOchreApplianceTeardown_ConfirmedPrintsSuccess(t *testing.T) { } func TestRunOchreApplianceTeardown_TeardownErrorExits1(t *testing.T) { - withApplianceTeardown(t, func(context.Context) error { + withApplianceTeardown(t, func(context.Context, bool) error { return errors.New("ochrevector: platform appliance is not enabled or not up on this node") }) @@ -102,3 +133,21 @@ func TestRunOchreApplianceTeardown_TeardownErrorExits1(t *testing.T) { code := withOchreExitCapture(t, func() { runOchreApplianceTeardown(&cmd, nil) }) require.Equal(t, 1, code) } + +// TestRunOchreApplianceTeardown_PurgeMetadataFlagReachesTeardownFn proves +// the --purge-metadata flag is read and passed through the Run wrapper. +func TestRunOchreApplianceTeardown_PurgeMetadataFlagReachesTeardownFn(t *testing.T) { + var gotPurge bool + withApplianceTeardown(t, func(_ context.Context, purgeMetadata bool) error { + gotPurge = purgeMetadata + return nil + }) + + cmd := *ochreApplianceTeardownCmd + require.NoError(t, cmd.Flags().Set("confirm", "true")) + require.NoError(t, cmd.Flags().Set("purge-metadata", "true")) + + code := withOchreExitCapture(t, func() { runOchreApplianceTeardown(&cmd, nil) }) + require.Equal(t, -1, code) + require.True(t, gotPurge, "--purge-metadata must reach applianceTeardownFn") +} diff --git a/spinifex/daemon/ochre_deps.go b/spinifex/daemon/ochre_deps.go index 57fff183a..bb317dbe4 100644 --- a/spinifex/daemon/ochre_deps.go +++ b/spinifex/daemon/ochre_deps.go @@ -189,10 +189,62 @@ func (d *Daemon) startOchreVector() { // (there is none today beyond observability) must never see a non-nil // service whose subjects are not yet actually serving. d.ochreVectorService = vectorService + + // Reconcile before the scheduler starts sweeping, so a job this pass + // enqueues for a rebuilt index is already PENDING for the scheduler's + // first Sweep rather than waiting for its next tick. + d.reconcileOchreIndexes(d.ctx, registry, backend, ingest) + go d.runOchreIngestScheduler(ingest) slog.Info("Ochre vector store enabled") } +// reconcileOchreIndexes re-creates and re-ingests every registry index whose +// backing Postgres table is missing -- the state left by a teardown that +// preserved metadata (registry/KB/DataSource) but destroyed the RDS +// instance, once Ensure has connected to a freshly-provisioned replacement. +// An index whose table survived (a plain restart) is left alone: no +// EnsureAccount/CreateIndex call and no ingest job for it. Best-effort: a +// failure on one record is logged and the pass continues to the rest rather +// than failing the appliance connect that already succeeded. +func (d *Daemon) reconcileOchreIndexes(ctx context.Context, registry *handlers_ochrevector.Registry, backend handlers_ochrevector.VectorBackend, ingest *handlers_ochrevector.IngestService) { + recs, err := registry.ListAll(ctx) + if err != nil { + slog.Warn("Ochre vector store: reconcile indexes: list registry failed", "err", err) + return + } + for _, rec := range recs { + exists, err := backend.IndexExists(ctx, rec.AccountID, rec.ID) + if err != nil { + slog.Warn("Ochre vector store: reconcile indexes: check exists failed", + "account", rec.AccountID, "index", rec.ID, "err", err) + continue + } + if exists { + continue + } + if err := backend.EnsureAccount(ctx, rec.AccountID); err != nil { + slog.Warn("Ochre vector store: reconcile indexes: ensure account failed", + "account", rec.AccountID, "index", rec.ID, "err", err) + continue + } + spec := handlers_ochrevector.IndexSpec{ID: rec.ID, Dimension: rec.Dimension} + if err := backend.CreateIndex(ctx, rec.AccountID, spec); err != nil { + slog.Warn("Ochre vector store: reconcile indexes: create index failed", + "account", rec.AccountID, "index", rec.ID, "err", err) + continue + } + for _, source := range rec.SourceSpecs { + if _, err := ingest.ReconcileFromSource(ctx, rec.AccountID, rec.ID, source); err != nil { + slog.Warn("Ochre vector store: reconcile indexes: enqueue ingest failed", + "account", rec.AccountID, "index", rec.ID, "err", err) + } + } + slog.Info("Ochre vector store: reconcile indexes: re-created index absent from fresh appliance", + "account", rec.AccountID, "index", rec.ID, "sources", len(rec.SourceSpecs)) + } +} + // ochreIngestSweepInterval paces the scheduler tick that drives PENDING ingest // jobs and re-drives stale RUNNING ones. const ochreIngestSweepInterval = 15 * time.Second @@ -225,7 +277,7 @@ func (d *Daemon) runOchreIngestScheduler(ingest *handlers_ochrevector.IngestServ // an earlier call) has nothing to act on, which is reported as an error // rather than silently accepted -- an operator asking to tear down expects // one to have existed. -func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, _ *handlers_ochrevector.TeardownApplianceRequest, _ string) (*handlers_ochrevector.TeardownApplianceResponse, error) { +func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, req *handlers_ochrevector.TeardownApplianceRequest, _ string) (*handlers_ochrevector.TeardownApplianceResponse, error) { d.mu.Lock() appliance := d.ochreAppliance d.mu.Unlock() @@ -233,7 +285,8 @@ func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, _ *handlers_o return nil, errors.New("ochrevector: platform appliance is not enabled or not up on this node") } - if err := appliance.Teardown(ctx); err != nil { + purgeMetadata := req != nil && req.PurgeMetadata + if err := appliance.Teardown(ctx, purgeMetadata); err != nil { return nil, fmt.Errorf("ochrevector: teardown platform appliance: %w", err) } diff --git a/spinifex/daemon/ochre_deps_test.go b/spinifex/daemon/ochre_deps_test.go index fcb9d5859..7884373df 100644 --- a/spinifex/daemon/ochre_deps_test.go +++ b/spinifex/daemon/ochre_deps_test.go @@ -7,15 +7,104 @@ package daemon import ( "context" "errors" + "sync" "testing" "time" "github.com/mulgadc/spinifex/spinifex/config" handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector" + "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// reconcileFakeBackend is a minimal handlers_ochrevector.VectorBackend +// double for reconcileOchreIndexes tests: only IndexExists/EnsureAccount/ +// CreateIndex are exercised, so every other method is an unused stub. +type reconcileFakeBackend struct { + mu sync.Mutex + indexes map[string]bool // key: accountID+"/"+indexID + + ensureAccountCalls []string + createIndexCalls []handlers_ochrevector.IndexSpec + + // Error-injection fields for the reconcile error-branch tests below: each + // makes the matching backend method fail on every call, so a test that + // sets one exercises exactly the corresponding continue+slog.Warn branch. + indexExistsErr error + ensureAccountErr error + createIndexErr error + + // onCreateIndex runs after a successful CreateIndex, letting a test mutate + // shared state (e.g. delete the registry record) without any production + // code change, so a later ReconcileFromSource call in the same pass fails. + onCreateIndex func(ctx context.Context, accountID string, spec handlers_ochrevector.IndexSpec) +} + +var _ handlers_ochrevector.VectorBackend = (*reconcileFakeBackend)(nil) + +func newReconcileFakeBackend() *reconcileFakeBackend { + return &reconcileFakeBackend{indexes: map[string]bool{}} +} + +func (f *reconcileFakeBackend) Init(_ context.Context) error { return nil } + +func (f *reconcileFakeBackend) EnsureAccount(_ context.Context, accountID string) error { + f.mu.Lock() + f.ensureAccountCalls = append(f.ensureAccountCalls, accountID) + err := f.ensureAccountErr + f.mu.Unlock() + return err +} + +func (f *reconcileFakeBackend) CreateIndex(ctx context.Context, accountID string, spec handlers_ochrevector.IndexSpec) error { + f.mu.Lock() + f.createIndexCalls = append(f.createIndexCalls, spec) + err := f.createIndexErr + if err == nil { + f.indexes[accountID+"/"+spec.ID] = true + } + onCreateIndex := f.onCreateIndex + f.mu.Unlock() + if err != nil { + return err + } + if onCreateIndex != nil { + onCreateIndex(ctx, accountID, spec) + } + return nil +} + +func (f *reconcileFakeBackend) IndexExists(_ context.Context, accountID, indexID string) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.indexExistsErr != nil { + return false, f.indexExistsErr + } + return f.indexes[accountID+"/"+indexID], nil +} + +func (f *reconcileFakeBackend) DropIndex(_ context.Context, accountID, indexID string) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.indexes, accountID+"/"+indexID) + return nil +} + +func (f *reconcileFakeBackend) ReplaceDocument(_ context.Context, _, _, _ string, _ []handlers_ochrevector.VectorRow) error { + return nil +} + +func (f *reconcileFakeBackend) Query(_ context.Context, _, _ string, _ []float32, _ int, _ *handlers_ochrevector.Filter) ([]handlers_ochrevector.QueryResult, error) { + return nil, nil +} + +func (f *reconcileFakeBackend) ensureAccountCallCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.ensureAccountCalls) +} + // TestStartOchreVector_DisabledSkipsConstruction pins the config gate's // default-off behavior (D-series Stage 5b): with OchreVector.Enabled false // (the zero value, i.e. every daemon that has not opted in), startOchreVector @@ -90,3 +179,175 @@ func TestHandleOchreApplianceTeardown_NilApplianceRefuses(t *testing.T) { assert.Nil(t, out) assert.Contains(t, err.Error(), "not enabled or not up") } + +const reconcileTestAccount = "111111111111" + +// TestReconcileOchreIndexes_RecreatesMissingIndexAndEnqueuesJob proves the +// reconcile pass re-creates a registry index whose pg table is absent (the +// state a fresh, teardown-and-rebuilt appliance leaves) and enqueues exactly +// one PENDING ingest job per persisted SourceSpec. +func TestReconcileOchreIndexes_RecreatesMissingIndexAndEnqueuesJob(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + registry := handlers_ochrevector.NewRegistry(js) + jobs := handlers_ochrevector.NewJobStore(js) + backend := newReconcileFakeBackend() + ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil) + + ctx := context.Background() + now := time.Now().UTC() + source := handlers_ochrevector.SourceSpec{Bucket: "docs", Prefix: "kb/", ChunkSize: 100, ChunkOverlap: 10, EmbeddingModel: "stub-embed", Dimension: 2} + require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, handlers_ochrevector.Record{ + ID: "idx-one", Dimension: 2, State: handlers_ochrevector.StateReady, + SourceSpecs: []handlers_ochrevector.SourceSpec{source}, CreatedAt: now, UpdatedAt: now, + })) + + d := &Daemon{} + d.reconcileOchreIndexes(ctx, registry, backend, ingest) + + assert.Equal(t, 1, backend.ensureAccountCallCount(), "must EnsureAccount before CreateIndex") + require.Len(t, backend.createIndexCalls, 1) + assert.Equal(t, "idx-one", backend.createIndexCalls[0].ID) + + jobRecs, err := jobs.ListAll(ctx) + require.NoError(t, err) + require.Len(t, jobRecs, 1, "must enqueue exactly one job for the one persisted SourceSpec") + assert.Equal(t, handlers_ochrevector.JobStatePending, jobRecs[0].State) + assert.Equal(t, source, jobRecs[0].Source) +} + +// TestReconcileOchreIndexes_ExistingIndexIsNoOp proves an index whose pg +// table already exists (a plain restart, not a teardown) is left alone: no +// EnsureAccount/CreateIndex call and no ingest job enqueued for it. +func TestReconcileOchreIndexes_ExistingIndexIsNoOp(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + registry := handlers_ochrevector.NewRegistry(js) + jobs := handlers_ochrevector.NewJobStore(js) + backend := newReconcileFakeBackend() + ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil) + + ctx := context.Background() + now := time.Now().UTC() + source := handlers_ochrevector.SourceSpec{Bucket: "docs", Prefix: "kb/", Dimension: 2} + require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, handlers_ochrevector.Record{ + ID: "idx-one", Dimension: 2, State: handlers_ochrevector.StateReady, + SourceSpecs: []handlers_ochrevector.SourceSpec{source}, CreatedAt: now, UpdatedAt: now, + })) + // Pre-seed the backend as if the table already survived a restart. + require.NoError(t, backend.CreateIndex(ctx, reconcileTestAccount, handlers_ochrevector.IndexSpec{ID: "idx-one", Dimension: 2})) + backend.mu.Lock() + backend.createIndexCalls = nil // reset so the assertion below is reconcile's own calls, not the seed + backend.mu.Unlock() + + d := &Daemon{} + d.reconcileOchreIndexes(ctx, registry, backend, ingest) + + assert.Zero(t, backend.ensureAccountCallCount(), "an already-present index must not be re-provisioned") + assert.Empty(t, backend.createIndexCalls) + + jobRecs, err := jobs.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, jobRecs, "an already-present index must not get a re-ingest job") +} + +// TestReconcileOchreIndexes_ErrorsAreLoggedAndSkipped proves each best-effort +// error branch (IndexExists, EnsureAccount, CreateIndex, ReconcileFromSource) +// logs and skips its own record without panicking, rather than failing the +// whole reconcile pass. +func TestReconcileOchreIndexes_ErrorsAreLoggedAndSkipped(t *testing.T) { + newRecord := func(id string) handlers_ochrevector.Record { + now := time.Now().UTC() + return handlers_ochrevector.Record{ + ID: id, Dimension: 2, State: handlers_ochrevector.StateReady, + SourceSpecs: []handlers_ochrevector.SourceSpec{{Bucket: "docs", Dimension: 2}}, + CreatedAt: now, UpdatedAt: now, + } + } + + t.Run("IndexExists error skips the record", func(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + registry := handlers_ochrevector.NewRegistry(js) + jobs := handlers_ochrevector.NewJobStore(js) + backend := newReconcileFakeBackend() + backend.indexExistsErr = errors.New("index exists boom") + ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil) + + ctx := context.Background() + require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-exists-err"))) + + d := &Daemon{} + assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) }) + + assert.Zero(t, backend.ensureAccountCallCount(), "IndexExists error must skip EnsureAccount") + assert.Empty(t, backend.createIndexCalls, "IndexExists error must skip CreateIndex") + jobRecs, err := jobs.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, jobRecs, "IndexExists error must skip the ingest enqueue") + }) + + t.Run("EnsureAccount error skips the record", func(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + registry := handlers_ochrevector.NewRegistry(js) + jobs := handlers_ochrevector.NewJobStore(js) + backend := newReconcileFakeBackend() + backend.ensureAccountErr = errors.New("ensure account boom") + ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil) + + ctx := context.Background() + require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-ensure-err"))) + + d := &Daemon{} + assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) }) + + assert.Equal(t, 1, backend.ensureAccountCallCount(), "EnsureAccount must still be attempted") + assert.Empty(t, backend.createIndexCalls, "EnsureAccount error must skip CreateIndex") + jobRecs, err := jobs.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, jobRecs, "EnsureAccount error must skip the ingest enqueue") + }) + + t.Run("CreateIndex error skips the record", func(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + registry := handlers_ochrevector.NewRegistry(js) + jobs := handlers_ochrevector.NewJobStore(js) + backend := newReconcileFakeBackend() + backend.createIndexErr = errors.New("create index boom") + ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil) + + ctx := context.Background() + require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-create-err"))) + + d := &Daemon{} + assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) }) + + require.Len(t, backend.createIndexCalls, 1, "CreateIndex must still be attempted") + jobRecs, err := jobs.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, jobRecs, "CreateIndex error must skip the ingest enqueue") + }) + + t.Run("ReconcileFromSource error is logged and does not panic", func(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + registry := handlers_ochrevector.NewRegistry(js) + jobs := handlers_ochrevector.NewJobStore(js) + backend := newReconcileFakeBackend() + ctx := context.Background() + // Delete the just-created record as a side effect of CreateIndex + // succeeding, so the ReconcileFromSource call right after it misses + // the registry and returns ErrIndexNotFound -- no production code + // change needed to force that failure. + backend.onCreateIndex = func(ctx context.Context, accountID string, spec handlers_ochrevector.IndexSpec) { + require.NoError(t, registry.Delete(ctx, accountID, spec.ID)) + } + ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil) + + require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-ingest-err"))) + + d := &Daemon{} + assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) }) + + require.Len(t, backend.createIndexCalls, 1, "CreateIndex must still run before the ingest enqueue fails") + jobRecs, err := jobs.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, jobRecs, "ReconcileFromSource error must not leave a job behind") + }) +} diff --git a/spinifex/handlers/ochrevector/appliance.go b/spinifex/handlers/ochrevector/appliance.go index 56625a64d..f343c83a5 100644 --- a/spinifex/handlers/ochrevector/appliance.go +++ b/spinifex/handlers/ochrevector/appliance.go @@ -92,9 +92,14 @@ type ApplianceRecord struct { UpdatedAt time.Time `json:"updatedAt"` } -// TeardownApplianceRequest has no fields: the appliance is a fixed, -// deployment-wide singleton (D2), never identified by a caller-supplied ID. -type TeardownApplianceRequest struct{} +// TeardownApplianceRequest identifies no resource -- the appliance is a +// fixed, deployment-wide singleton (D2). PurgeMetadata opts into the old +// full wipe (index registry, KB and DataSource records alongside the RDS +// instance); the default leaves that metadata intact so a later Ensure can +// reconcile and re-ingest from it without operator input. +type TeardownApplianceRequest struct { + PurgeMetadata bool `json:"purgeMetadata,omitempty"` +} type TeardownApplianceResponse struct{} @@ -215,14 +220,17 @@ func (a *Appliance) TeardownHostPort() error { } // Teardown removes the singleton platform appliance -- RDS instance, host -// port, then (if WithStores/WithKBStores were called) index/job and -// KB/DataSource metadata, then the KV record -- so a rebuilt appliance starts -// coherent. Every step runs -// regardless of an earlier one's failure, and every failure is joined into -// the returned error. Idempotent overall: tearing down an already-absent -// appliance is a no-op success. Does not re-provision -- the daemon's own -// Ensure does that on next startup, not this call. -func (a *Appliance) Teardown(ctx context.Context) error { +// port, then the appliance KV record -- so a rebuilt appliance starts +// coherent. Index/job registry and KB/DataSource metadata (if +// WithStores/WithKBStores were called) survive by default exactly as they +// survive a restart, so a later Ensure can reconcile and re-ingest from them +// without operator input; purgeMetadata=true restores the old full wipe of +// all of it for an intentional destroy. Every step runs regardless of an +// earlier one's failure, and every failure is joined into the returned +// error. Idempotent overall: tearing down an already-absent appliance is a +// no-op success. Does not re-provision -- the daemon's own Ensure does that +// on next startup, not this call. +func (a *Appliance) Teardown(ctx context.Context, purgeMetadata bool) error { var errs []error if err := a.launcher.Delete(ctx, ApplianceIdentifier); err != nil { @@ -237,30 +245,34 @@ func (a *Appliance) Teardown(ctx context.Context) error { a.mu.Lock() registry, jobs, kb, ds := a.registry, a.jobs, a.kb, a.ds a.mu.Unlock() - // Purged after the data-bearing RDS instance is gone but before the - // appliance record: a crash here still leaves a record an operator can - // retry teardown against, re-attempting the (idempotent) purge too. - if registry != nil { - if err := registry.PurgeAll(ctx); err != nil { - errs = append(errs, fmt.Errorf("ochrevector: purge index registry: %w", err)) - } - } + // Job history is scoped to the now-destroyed RDS instance, so it is + // always cleared. Registry/kb/ds instead carry the identity and + // SourceSpecs a fresh Ensure needs to reconcile and re-ingest without + // operator input, so they purge only when purgeMetadata asks for the old + // full wipe. if jobs != nil { if err := jobs.PurgeAll(ctx); err != nil { errs = append(errs, fmt.Errorf("ochrevector: purge ingestion jobs: %w", err)) } } - // kb/ds are gateway-owned metadata, not daemon-owned like registry/jobs - // above, but they still point at this appliance's index/backend, so a - // torn-down appliance must not leave them dangling either. - if kb != nil { - if err := kb.PurgeAll(ctx); err != nil { - errs = append(errs, fmt.Errorf("ochrevector: purge knowledge bases: %w", err)) + if purgeMetadata { + if registry != nil { + if err := registry.PurgeAll(ctx); err != nil { + errs = append(errs, fmt.Errorf("ochrevector: purge index registry: %w", err)) + } } - } - if ds != nil { - if err := ds.PurgeAll(ctx); err != nil { - errs = append(errs, fmt.Errorf("ochrevector: purge data sources: %w", err)) + // kb/ds are gateway-owned metadata, not daemon-owned like + // registry/jobs above, but they still point at this appliance's + // index/backend, so a full wipe must not leave them dangling either. + if kb != nil { + if err := kb.PurgeAll(ctx); err != nil { + errs = append(errs, fmt.Errorf("ochrevector: purge knowledge bases: %w", err)) + } + } + if ds != nil { + if err := ds.PurgeAll(ctx); err != nil { + errs = append(errs, fmt.Errorf("ochrevector: purge data sources: %w", err)) + } } } diff --git a/spinifex/handlers/ochrevector/appliance_test.go b/spinifex/handlers/ochrevector/appliance_test.go index c9e674538..e8952a3f7 100644 --- a/spinifex/handlers/ochrevector/appliance_test.go +++ b/spinifex/handlers/ochrevector/appliance_test.go @@ -519,7 +519,7 @@ func TestTeardown_DeletesRecordLauncherAndHostPort(t *testing.T) { appliance.eniID = "eni-installed" appliance.mu.Unlock() - require.NoError(t, appliance.Teardown(context.Background())) + require.NoError(t, appliance.Teardown(context.Background(), false)) assert.Equal(t, []string{ApplianceIdentifier}, launcher.deleteCalls) assert.Equal(t, []string{"eni-installed"}, h.hostPort.removals()) @@ -540,11 +540,11 @@ func TestTeardown_NoExistingApplianceIsANoOp(t *testing.T) { appliance, err := NewAppliance(js, testMasterKey(t), launcher) require.NoError(t, err) - assert.NoError(t, appliance.Teardown(context.Background())) + assert.NoError(t, appliance.Teardown(context.Background(), false)) assert.Equal(t, 1, launcher.deleteCallCount()) // Tearing down again must still be a no-op success. - assert.NoError(t, appliance.Teardown(context.Background())) + assert.NoError(t, appliance.Teardown(context.Background(), false)) } // TestTeardown_StillDeletesRecordWhenHostPortRemovalFails proves a host-port @@ -565,7 +565,7 @@ func TestTeardown_StillDeletesRecordWhenHostPortRemovalFails(t *testing.T) { appliance.eniID = "eni-installed" appliance.mu.Unlock() - err = appliance.Teardown(context.Background()) + err = appliance.Teardown(context.Background(), false) require.Error(t, err) assert.Contains(t, err.Error(), "ovs-vsctl exploded") @@ -587,7 +587,7 @@ func TestTeardown_JoinsLauncherAndRecordErrors(t *testing.T) { require.NoError(t, err) seedAvailableAppliance(t, appliance, masterKey, "10.0.0.32", 5432) - err = appliance.Teardown(context.Background()) + err = appliance.Teardown(context.Background(), false) require.Error(t, err) assert.Contains(t, err.Error(), "rds unreachable") @@ -598,11 +598,11 @@ func TestTeardown_JoinsLauncherAndRecordErrors(t *testing.T) { assert.Nil(t, rec, "the KV record must still be deleted despite the launcher failure") } -// TestTeardown_WithStoresPurgesRegistryAndJobs proves an Appliance wired via -// WithStores purges every index and job record on Teardown, so a -// reprovisioned appliance never advertises metadata pointing at data that no -// longer exists. -func TestTeardown_WithStoresPurgesRegistryAndJobs(t *testing.T) { +// TestTeardown_DefaultPreservesRegistryButPurgesJobs proves a default +// Teardown (purgeMetadata=false) leaves the index registry intact -- so a +// later Ensure can reconcile from it -- while still clearing job history, +// which is scoped to the now-destroyed RDS instance. +func TestTeardown_DefaultPreservesRegistryButPurgesJobs(t *testing.T) { _, _, js := testutil.StartTestJetStream(t) masterKey := testMasterKey(t) launcher := &fakeLauncher{endpoint: "10.0.0.33", port: 5432} @@ -618,22 +618,51 @@ func TestTeardown_WithStoresPurgesRegistryAndJobs(t *testing.T) { require.NoError(t, jobs.Reserve(ctx, "111111111111", JobRecord{ID: "job-one", CreatedAt: now, UpdatedAt: now})) appliance.WithStores(registry, jobs) - require.NoError(t, appliance.Teardown(ctx)) + require.NoError(t, appliance.Teardown(ctx, false)) remainingIndexes, err := registry.ListAll(ctx) require.NoError(t, err) - assert.Empty(t, remainingIndexes, "Teardown must purge every index record") + assert.Len(t, remainingIndexes, 1, "default Teardown must leave the index registry intact") remainingJobs, err := jobs.ListAll(ctx) require.NoError(t, err) - assert.Empty(t, remainingJobs, "Teardown must purge every job record") + assert.Empty(t, remainingJobs, "Teardown must still purge job history") } -// TestTeardown_WithKBStoresPurgesKnowledgeBasesAndDataSources proves an -// Appliance wired via WithKBStores purges every gateway-owned knowledge-base -// and data-source record on Teardown, so a reprovisioned appliance never -// leaves KB/DataSource metadata pointing at an index that no longer exists. -func TestTeardown_WithKBStoresPurgesKnowledgeBasesAndDataSources(t *testing.T) { +// TestTeardown_PurgeMetadataPurgesRegistryAndJobs proves purgeMetadata=true +// restores the old full wipe of the index registry (and job history) +// alongside the RDS instance, for an intentional full destroy. +func TestTeardown_PurgeMetadataPurgesRegistryAndJobs(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + masterKey := testMasterKey(t) + launcher := &fakeLauncher{endpoint: "10.0.0.36", port: 5432} + appliance, err := NewAppliance(js, masterKey, launcher) + require.NoError(t, err) + seedAvailableAppliance(t, appliance, masterKey, "10.0.0.36", 5432) + + registry := NewRegistry(js) + jobs := NewJobStore(js) + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, registry.Reserve(ctx, "111111111111", Record{ID: "idx-one", CreatedAt: now, UpdatedAt: now})) + require.NoError(t, jobs.Reserve(ctx, "111111111111", JobRecord{ID: "job-one", CreatedAt: now, UpdatedAt: now})) + + appliance.WithStores(registry, jobs) + require.NoError(t, appliance.Teardown(ctx, true)) + + remainingIndexes, err := registry.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, remainingIndexes, "purgeMetadata=true must purge every index record") + + remainingJobs, err := jobs.ListAll(ctx) + require.NoError(t, err) + assert.Empty(t, remainingJobs, "purgeMetadata=true must still purge every job record") +} + +// TestTeardown_DefaultPreservesKnowledgeBasesAndDataSources proves a default +// Teardown leaves gateway-owned KB/DataSource records intact, mirroring the +// registry: their SourceSpec is what a later reconcile re-ingests from. +func TestTeardown_DefaultPreservesKnowledgeBasesAndDataSources(t *testing.T) { _, _, js := testutil.StartTestJetStream(t) masterKey := testMasterKey(t) launcher := &fakeLauncher{endpoint: "10.0.0.35", port: 5432} @@ -649,15 +678,80 @@ func TestTeardown_WithKBStoresPurgesKnowledgeBasesAndDataSources(t *testing.T) { require.NoError(t, ds.Create(ctx, "111111111111", DataSourceRecord{ID: "ds-one", KnowledgeBaseID: "kb-one", CreatedAt: now, UpdatedAt: now})) appliance.WithKBStores(kb, ds) - require.NoError(t, appliance.Teardown(ctx)) + require.NoError(t, appliance.Teardown(ctx, false)) remainingKBs, err := kb.List(ctx, "111111111111") require.NoError(t, err) - assert.Empty(t, remainingKBs, "Teardown must purge every knowledge base record") + assert.Len(t, remainingKBs, 1, "default Teardown must leave knowledge base records intact") remainingDS, err := ds.List(ctx, "111111111111") require.NoError(t, err) - assert.Empty(t, remainingDS, "Teardown must purge every data source record") + assert.Len(t, remainingDS, 1, "default Teardown must leave data source records intact") +} + +// TestTeardown_PurgeMetadataPurgesKnowledgeBasesAndDataSources proves +// purgeMetadata=true restores the old full wipe of the gateway-owned +// knowledge-base and data-source records. +func TestTeardown_PurgeMetadataPurgesKnowledgeBasesAndDataSources(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + masterKey := testMasterKey(t) + launcher := &fakeLauncher{endpoint: "10.0.0.37", port: 5432} + appliance, err := NewAppliance(js, masterKey, launcher) + require.NoError(t, err) + seedAvailableAppliance(t, appliance, masterKey, "10.0.0.37", 5432) + + kb := NewKBStore(js) + ds := NewDataSourceStore(js) + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, kb.Create(ctx, "111111111111", KBRecord{ID: "kb-one", IndexID: "idx-one", CreatedAt: now, UpdatedAt: now})) + require.NoError(t, ds.Create(ctx, "111111111111", DataSourceRecord{ID: "ds-one", KnowledgeBaseID: "kb-one", CreatedAt: now, UpdatedAt: now})) + + appliance.WithKBStores(kb, ds) + require.NoError(t, appliance.Teardown(ctx, true)) + + remainingKBs, err := kb.List(ctx, "111111111111") + require.NoError(t, err) + assert.Empty(t, remainingKBs, "purgeMetadata=true must purge every knowledge base record") + + remainingDS, err := ds.List(ctx, "111111111111") + require.NoError(t, err) + assert.Empty(t, remainingDS, "purgeMetadata=true must purge every data source record") +} + +// TestTeardown_PurgeMetadataSurfacesStorePurgeErrors proves a purgeMetadata= +// true Teardown reports (rather than swallows) a failure from any of +// registry/kb/ds PurgeAll, joined alongside every other step's own errors. +func TestTeardown_PurgeMetadataSurfacesStorePurgeErrors(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + masterKey := testMasterKey(t) + launcher := &fakeLauncher{endpoint: "10.0.0.39", port: 5432} + appliance, err := NewAppliance(js, masterKey, launcher) + require.NoError(t, err) + seedAvailableAppliance(t, appliance, masterKey, "10.0.0.39", 5432) + + registry := NewRegistry(js) + jobs := NewJobStore(js) + kb := NewKBStore(js) + ds := NewDataSourceStore(js) + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, registry.Reserve(ctx, "111111111111", Record{ID: "idx-one", CreatedAt: now, UpdatedAt: now})) + require.NoError(t, kb.Create(ctx, "111111111111", KBRecord{ID: "kb-one", IndexID: "idx-one", CreatedAt: now, UpdatedAt: now})) + require.NoError(t, ds.Create(ctx, "111111111111", DataSourceRecord{ID: "ds-one", KnowledgeBaseID: "kb-one", CreatedAt: now, UpdatedAt: now})) + + appliance.WithStores(registry, jobs) + appliance.WithKBStores(kb, ds) + + // Sever the shared NATS connection so every store's PurgeAll call fails, + // without any production code change. + nc.Close() + + err = appliance.Teardown(ctx, true) + require.Error(t, err) + assert.Contains(t, err.Error(), "purge index registry") + assert.Contains(t, err.Error(), "purge knowledge bases") + assert.Contains(t, err.Error(), "purge data sources") } // TestTeardown_WithoutStoresStillWorks proves an Appliance that never calls @@ -671,7 +765,7 @@ func TestTeardown_WithoutStoresStillWorks(t *testing.T) { require.NoError(t, err) seedAvailableAppliance(t, appliance, masterKey, "10.0.0.34", 5432) - require.NoError(t, appliance.Teardown(context.Background())) + require.NoError(t, appliance.Teardown(context.Background(), false)) } // TestBucket_RequiresJetStream proves bucket fails fast on a zero-value diff --git a/spinifex/handlers/ochrevector/backend.go b/spinifex/handlers/ochrevector/backend.go index 072884f7a..0bbd16033 100644 --- a/spinifex/handlers/ochrevector/backend.go +++ b/spinifex/handlers/ochrevector/backend.go @@ -71,6 +71,12 @@ type VectorBackend interface { // accountID's schema. Idempotent: safe to retry after a crash mid-create. CreateIndex(ctx context.Context, accountID string, spec IndexSpec) error + // IndexExists reports whether indexID's backing table already exists + // under accountID's schema, so a caller can tell a survived-restart index + // from one whose Postgres was rebuilt out from under it without + // re-creating (and re-ingesting) one that is already there. + IndexExists(ctx context.Context, accountID, indexID string) (bool, error) + // DropIndex drops indexID's backing table under accountID's schema. // Idempotent: dropping an already-absent index is a no-op success. DropIndex(ctx context.Context, accountID, indexID string) error diff --git a/spinifex/handlers/ochrevector/fake_backend_test.go b/spinifex/handlers/ochrevector/fake_backend_test.go index 4a6ae13e4..0bf5549ee 100644 --- a/spinifex/handlers/ochrevector/fake_backend_test.go +++ b/spinifex/handlers/ochrevector/fake_backend_test.go @@ -65,6 +65,13 @@ func (f *fakeBackend) CreateIndex(_ context.Context, accountID string, spec Inde return nil } +// IndexExists mirrors pgxBackend.IndexExists against the same indexes map +// CreateIndex/DropIndex mutate, so it reflects exactly what a real backend +// would report without a database. +func (f *fakeBackend) IndexExists(_ context.Context, accountID, indexID string) (bool, error) { + return f.hasIndex(accountID, indexID), nil +} + func (f *fakeBackend) DropIndex(_ context.Context, accountID, indexID string) error { f.mu.Lock() defer f.mu.Unlock() diff --git a/spinifex/handlers/ochrevector/ingest.go b/spinifex/handlers/ochrevector/ingest.go index 8654ed14c..7aa36dc91 100644 --- a/spinifex/handlers/ochrevector/ingest.go +++ b/spinifex/handlers/ochrevector/ingest.go @@ -107,6 +107,15 @@ func (s *IngestService) StartIngest(ctx context.Context, accountID, indexID stri return &rec, nil } +// ReconcileFromSource enqueues a PENDING ingest job for accountID/indexID +// from a persisted SourceSpec -- the entry point a startup reconcile pass +// calls to re-populate an index whose backing table survived only in the +// registry, not the freshly-provisioned Postgres. StartIngest's own +// reservation makes a repeat call safe: Sweep/RunJob never double-write. +func (s *IngestService) ReconcileFromSource(ctx context.Context, accountID, indexID string, source SourceSpec) (*JobRecord, error) { + return s.StartIngest(ctx, accountID, indexID, source, "") +} + // RunJob is the synchronous ingestion worker: lists job.Source, and for each // object, reads/chunks/embeds/replaces its rows. A per-document failure // (unreadable object, oversize, exhausted embed retries) is skipped and diff --git a/spinifex/handlers/ochrevector/ingest_test.go b/spinifex/handlers/ochrevector/ingest_test.go index f50d2f4bd..1642b5044 100644 --- a/spinifex/handlers/ochrevector/ingest_test.go +++ b/spinifex/handlers/ochrevector/ingest_test.go @@ -218,6 +218,34 @@ func TestStartIngest_StampsDataSourceID(t *testing.T) { assert.Empty(t, noDSJob.DataSourceID) } +// TestReconcileFromSource_EnqueuesPendingJobWithNoDataSourceID proves the +// startup-reconcile entry point enqueues a PENDING job from a bare +// SourceSpec, tagged with no DataSourceID -- it re-populates the index's +// table directly from the registry, not through a bedrock-agent DataSource. +func TestReconcileFromSource_EnqueuesPendingJobWithNoDataSourceID(t *testing.T) { + svc, _, _, _, _ := newIngestTestSetup(t) + ctx := context.Background() + + job, err := svc.ReconcileFromSource(ctx, ingestAccountA, "idx-one", testSource()) + require.NoError(t, err) + require.NotNil(t, job) + assert.Equal(t, JobStatePending, job.State) + assert.Empty(t, job.DataSourceID) + assert.Equal(t, testSource(), job.Source) +} + +// TestReconcileFromSource_MissingIndexErrors proves the entry point refuses +// the same way StartIngest does for an index the registry does not have -- +// a reconcile pass must never enqueue a job against nothing. +func TestReconcileFromSource_MissingIndexErrors(t *testing.T) { + svc, _, _, _, _ := newIngestTestSetup(t) + ctx := context.Background() + + _, err := svc.ReconcileFromSource(ctx, ingestAccountA, "idx-does-not-exist", testSource()) + require.Error(t, err) + assert.ErrorIs(t, err, ErrIndexNotFound) +} + func TestRunJob_HappyPath(t *testing.T) { svc, registry, backend, store, embedder := newIngestTestSetup(t) ctx := context.Background() diff --git a/spinifex/handlers/ochrevector/pgx_backend.go b/spinifex/handlers/ochrevector/pgx_backend.go index f7a402857..257e565be 100644 --- a/spinifex/handlers/ochrevector/pgx_backend.go +++ b/spinifex/handlers/ochrevector/pgx_backend.go @@ -223,6 +223,25 @@ func (b *pgxBackend) CreateIndex(ctx context.Context, accountID string, spec Ind }) } +// IndexExists reports whether indexID's backing table exists under +// accountID's schema. to_regclass resolves to NULL for a schema or table +// that is not there rather than erroring, so no account role or search_path +// setup is needed for this read-only lookup. +func (b *pgxBackend) IndexExists(ctx context.Context, accountID, indexID string) (bool, error) { + if err := validateAccountID(accountID); err != nil { + return false, err + } + if err := validateIndexID(indexID); err != nil { + return false, err + } + qualified := schemaName(accountID) + "." + tableName(indexID) + var regclass *string + if err := b.pool.QueryRow(ctx, `SELECT to_regclass($1)::text`, qualified).Scan(®class); err != nil { + return false, fmt.Errorf("ochrevector: check index %s exists for account %s: %w", indexID, accountID, err) + } + return regclass != nil, nil +} + // DropIndex drops indexID's backing table under accountID's schema. Idempotent: // dropping an already-absent index is a no-op success. func (b *pgxBackend) DropIndex(ctx context.Context, accountID, indexID string) error { diff --git a/spinifex/handlers/ochrevector/pgx_backend_test.go b/spinifex/handlers/ochrevector/pgx_backend_test.go index cb5e72a2b..8a7e75b8b 100644 --- a/spinifex/handlers/ochrevector/pgx_backend_test.go +++ b/spinifex/handlers/ochrevector/pgx_backend_test.go @@ -47,8 +47,16 @@ func TestPgxBackend_Integration(t *testing.T) { _ = backend.DropIndex(context.Background(), pgTestAccountA, indexID) }) + exists, err := backend.IndexExists(ctx, pgTestAccountA, indexID) + require.NoError(t, err) + assert.False(t, exists, "index must not exist before CreateIndex") + require.NoError(t, backend.CreateIndex(ctx, pgTestAccountA, IndexSpec{ID: indexID, Dimension: 8})) + exists, err = backend.IndexExists(ctx, pgTestAccountA, indexID) + require.NoError(t, err) + assert.True(t, exists, "index must exist after CreateIndex") + // Re-running CreateIndex must not error (IF NOT EXISTS), the property // Reconcile's forward-retry depends on. require.NoError(t, backend.CreateIndex(ctx, pgTestAccountA, IndexSpec{ID: indexID, Dimension: 8})) @@ -69,6 +77,10 @@ func TestPgxBackend_Integration(t *testing.T) { require.NoError(t, backend.DropIndex(ctx, pgTestAccountA, indexID)) + exists, err = backend.IndexExists(ctx, pgTestAccountA, indexID) + require.NoError(t, err) + assert.False(t, exists, "index must not exist after DropIndex") + // Dropping an already-absent index is a no-op success. require.NoError(t, backend.DropIndex(ctx, pgTestAccountA, indexID)) }