fix(indexjobs): deleting a catalog spec no longer strands index jobs in permanent Degraded

internal/platform/indexqueue/indexqueue_test.go

@@ -54,6 +54,8 @@ func (stubStore) Get(context.Context, int64) (*indexjobs.Job, error) {
 	return nil, indexjobs.ErrNotFound
 }
 
+func (stubStore) CancelPending(context.Context, indexjobs.Key) (int, error) { return 0, nil }
+
 func (stubStore) List(context.Context, indexjobs.ListFilter) ([]indexjobs.Job, error) {
 	return nil, nil
 }

internal/platform/indexqueue/sources.go

@@ -39,14 +39,18 @@ type catalogSource struct {
 func (*catalogSource) Kind() string { return catalogindex.SourceKind }
 
 // LoadItems decodes the source_id, fetches the spec content, and returns one
-// item per operation. A missing spec surfaces as an error (the worker treats it
-// as terminal: the spec was deleted).
+// item per operation. A missing spec (or catalog) surfaces as an error wrapping
+// indexjobs.ErrSourceGone, so the worker resolves the unit — the spec was
+// deleted, not broken — instead of recording an open failure (#998).
 func (s *catalogSource) LoadItems(ctx context.Context, sourceID string) ([]indexjobs.Item, error) {
 	catalogID, specName, ok := catalogindex.DecodeSourceID(sourceID)
 	if !ok {
 		return nil, fmt.Errorf("catalogSource: malformed source_id %q", sourceID)
 	}
 	spec, err := s.store.GetSpec(ctx, catalogID, specName)
+	if errors.Is(err, apigatewaycatalog.ErrNotFound) {
+		return nil, fmt.Errorf("catalogSource: %w: %w", indexjobs.ErrSourceGone, err)
+	}
 	if err != nil {
 		return nil, fmt.Errorf("catalogSource: get spec: %w", err)
 	}

internal/platform/indexqueue/sources_test.go

@@ -3,10 +3,12 @@ package indexqueue
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/modelcontextprotocol/go-sdk/mcp"
 
+	"github.com/txn2/mcp-data-platform/pkg/indexjobs"
 	"github.com/txn2/mcp-data-platform/pkg/registry"
 	apigatewaykit "github.com/txn2/mcp-data-platform/pkg/toolkits/apigateway"
 	apigatewaycatalog "github.com/txn2/mcp-data-platform/pkg/toolkits/apigateway/catalog"
@@ -98,15 +100,39 @@ func TestCatalogSource_LoadItems_MalformedSourceID(t *testing.T) {
 	}
 }
 
-// TestCatalogSource_LoadItems_MissingSpec proves a vanished spec surfaces as a
-// wrapped error; the worker treats it as terminal (the spec was deleted between
-// enqueue and claim).
+// TestCatalogSource_LoadItems_MissingSpec proves a vanished spec surfaces as an
+// error wrapping indexjobs.ErrSourceGone, so the worker resolves the unit
+// (the spec was deleted between enqueue and claim) instead of recording an
+// open failure nothing can ever supersede (#998).
 func TestCatalogSource_LoadItems_MissingSpec(t *testing.T) {
 	t.Parallel()
 	s := &catalogSource{store: apigatewaycatalog.NewMemoryStore()}
-	if _, err := s.LoadItems(context.Background(), catalogindex.EncodeSourceID("missing", "missing")); err == nil {
+	_, err := s.LoadItems(context.Background(), catalogindex.EncodeSourceID("missing", "missing"))
+	if err == nil {
 		t.Fatal("expected error for missing spec")
 	}
+	if !errors.Is(err, indexjobs.ErrSourceGone) {
+		t.Errorf("missing spec error = %v; must wrap indexjobs.ErrSourceGone", err)
+	}
+	if !strings.Contains(err.Error(), `"missing"`) {
+		t.Errorf("missing spec error = %v; must name the missing spec", err)
+	}
+}
+
+// TestCatalogSource_LoadItems_StoreErrorIsNotGone pins the boundary of the
+// gone signal: an unreadable store is NOT a deleted source, so the error must
+// not wrap ErrSourceGone (the worker would silently resolve a unit that still
+// exists).
+func TestCatalogSource_LoadItems_StoreErrorIsNotGone(t *testing.T) {
+	t.Parallel()
+	s := &catalogSource{store: &errStore{err: errors.New("db down")}}
+	_, err := s.LoadItems(context.Background(), catalogindex.EncodeSourceID("c", "s"))
+	if err == nil {
+		t.Fatal("expected store error to surface")
+	}
+	if errors.Is(err, indexjobs.ErrSourceGone) {
+		t.Errorf("store error = %v; must NOT wrap ErrSourceGone", err)
+	}
 }
 
 // TestCatalogSource_LoadItems_ParseError proves malformed spec content surfaces

internal/testdb/testdb.go

@@ -49,15 +49,22 @@ func NewWithDSN(t *testing.T) (*sql.DB, string) {
 	}
 	ctx := context.Background()
 
+	// The wait strategy requires BOTH the second "ready" log line (postgres
+	// restarts once during init) AND the mapped port to be externally
+	// reachable. The log line alone raced under parallel container load:
+	// ConnectionString could run before Docker exposed the port mapping and
+	// fail with `port "5432/tcp" not found`.
 	container, err := tcpostgres.Run(ctx,
 		"pgvector/pgvector:pg16",
 		tcpostgres.WithDatabase("testdb"),
 		tcpostgres.WithUsername("test"),
 		tcpostgres.WithPassword("test"),
 		testcontainers.WithWaitStrategy(
-			wait.ForLog("database system is ready to accept connections").
-				WithOccurrence(2).
-				WithStartupTimeout(5*time.Minute),
+			wait.ForAll(
+				wait.ForLog("database system is ready to accept connections").
+					WithOccurrence(2),
+				wait.ForListeningPort("5432/tcp"),
+			).WithStartupTimeoutDefault(5*time.Minute),
 		),
 	)
 	if err != nil {

pkg/admin/catalog_delete_realdb_integration_test.go

@@ -0,0 +1,154 @@
+//go:build integration
+
+package admin
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/txn2/mcp-data-platform/internal/testdb"
+	"github.com/txn2/mcp-data-platform/pkg/indexjobs"
+	apicatalog "github.com/txn2/mcp-data-platform/pkg/toolkits/apigateway/catalog"
+	"github.com/txn2/mcp-data-platform/pkg/toolkits/apigateway/catalogindex"
+)
+
+// specBody is a one-operation OpenAPI document the admin spec write
+// path accepts, so each PUT enqueues exactly one index job through the
+// real producer hook.
+const specBody = `openapi: 3.0.0
+info: {title: t, version: "1"}
+paths:
+  /a:
+    get:
+      operationId: a
+      responses:
+        "200":
+          description: ok`
+
+// requireNoResidue asserts the acceptance criteria of #998 for one
+// unit: zero pending/running jobs for the key, zero open failures, and
+// an api_catalog kind that reports no unresolved failures (the signal
+// the portal's Degraded verdict keys on).
+func requireNoResidue(t *testing.T, jobs *indexjobs.PostgresStore, catalogID, specName string) {
+	t.Helper()
+	ctx := context.Background()
+	sourceID := catalogindex.EncodeSourceID(catalogID, specName)
+
+	rows, err := jobs.List(ctx, indexjobs.ListFilter{SourceKind: catalogindex.SourceKind, SourceID: sourceID})
+	require.NoError(t, err)
+	for _, j := range rows {
+		require.NotEqual(t, indexjobs.StatusPending, j.Status, "pending job survived the delete: %+v", j)
+		require.NotEqual(t, indexjobs.StatusRunning, j.Status, "running job survived the delete: %+v", j)
+	}
+
+	failures, err := jobs.ActiveFailures(ctx, catalogindex.SourceKind, 50)
+	require.NoError(t, err)
+	for _, f := range failures {
+		require.NotEqual(t, sourceID, f.SourceID, "open failure survived the delete: %+v", f)
+	}
+
+	counts, err := jobs.Counts(ctx, catalogindex.SourceKind)
+	require.NoError(t, err)
+	require.Zero(t, counts.UnresolvedFailures, "api_catalog kind must report healthy after the delete")
+}
+
+// TestCatalogSpecDelete_ClearsIndexResidue_RealDB drives the real
+// admin delete handlers against the real Postgres-backed indexjobs
+// store (#998): deleting a spec (or its whole catalog) must leave zero
+// pending jobs and zero open failures for the affected keys, with no
+// operator dismiss.
+func TestCatalogSpecDelete_ClearsIndexResidue_RealDB(t *testing.T) {
+	ctx := context.Background()
+	db := testdb.New(t)
+
+	catalogStore := apicatalog.NewPostgresStore(db)
+	jobs := indexjobs.NewPostgresStore(db)
+	h := NewHandler(Deps{
+		APICatalogStore:   catalogStore,
+		EmbedJobs:         catalogindex.NewAdminStore(jobs, db),
+		ConfigStore:       &mockConfigStore{mode: "database"},
+		DatabaseAvailable: true,
+	}, nil)
+
+	require.NoError(t, catalogStore.CreateCatalog(ctx, apicatalog.Catalog{
+		ID: "cat1", Name: "cat1", DisplayName: "Catalog One",
+	}))
+
+	putSpec := func(catalogID, specName string) {
+		t.Helper()
+		res := doJSON(t, h, http.MethodPut, "/api/v1/admin/api-catalogs/"+catalogID+"/specs/"+specName, map[string]any{
+			"source_kind": "inline",
+			"content":     specBody,
+		})
+		require.Equal(t, http.StatusOK, res.Code, "upsert %s/%s: %s", catalogID, specName, res.Body.String())
+	}
+
+	t.Run("pending job is canceled by spec delete", func(t *testing.T) {
+		putSpec("cat1", "alpha")
+
+		pending, err := jobs.List(ctx, indexjobs.ListFilter{
+			SourceKind: catalogindex.SourceKind,
+			SourceID:   catalogindex.EncodeSourceID("cat1", "alpha"),
+			Status:     indexjobs.StatusPending,
+		})
+		require.NoError(t, err)
+		require.Len(t, pending, 1, "spec write must enqueue a pending job")
+
+		res := doJSON(t, h, http.MethodDelete, "/api/v1/admin/api-catalogs/cat1/specs/alpha", nil)
+		require.Equal(t, http.StatusOK, res.Code, res.Body.String())
+
+		requireNoResidue(t, jobs, "cat1", "alpha")
+	})
+
+	t.Run("open failure is resolved by spec delete", func(t *testing.T) {
+		putSpec("cat1", "beta")
+
+		job, err := jobs.Claim(ctx, "it-worker")
+		require.NoError(t, err)
+		require.Equal(t, catalogindex.EncodeSourceID("cat1", "beta"), job.SourceID)
+		require.NoError(t, jobs.Fail(ctx, job.ID, "it-worker", "catalogSource: get spec: boom"))
+
+		failures, err := jobs.ActiveFailures(ctx, catalogindex.SourceKind, 50)
+		require.NoError(t, err)
+		require.Len(t, failures, 1, "the failed job must surface as an open failure before the delete")
+
+		res := doJSON(t, h, http.MethodDelete, "/api/v1/admin/api-catalogs/cat1/specs/beta", nil)
+		require.Equal(t, http.StatusOK, res.Code, res.Body.String())
+
+		requireNoResidue(t, jobs, "cat1", "beta")
+	})
+
+	t.Run("catalog delete clears residue for every spec", func(t *testing.T) {
+		require.NoError(t, catalogStore.CreateCatalog(ctx, apicatalog.Catalog{
+			ID: "cat2", Name: "cat2", DisplayName: "Catalog Two",
+		}))
+		putSpec("cat2", "gamma")
+		putSpec("cat2", "delta")
+
+		// Leave gamma's job pending and fail delta's, covering both
+		// residue shapes in one catalog delete.
+		var deltaJob *indexjobs.Job
+		for {
+			job, err := jobs.Claim(ctx, "it-worker")
+			if err != nil {
+				break
+			}
+			if job.SourceID == catalogindex.EncodeSourceID("cat2", "delta") {
+				deltaJob = job
+			} else {
+				require.NoError(t, jobs.Retry(ctx, job.ID, "it-worker", "put back"))
+			}
+		}
+		require.NotNil(t, deltaJob, "delta job must be claimable")
+		require.NoError(t, jobs.Fail(ctx, deltaJob.ID, "it-worker", "catalogSource: get spec: boom"))
+
+		res := doJSON(t, h, http.MethodDelete, "/api/v1/admin/api-catalogs/cat2", nil)
+		require.Equal(t, http.StatusOK, res.Code, res.Body.String())
+
+		requireNoResidue(t, jobs, "cat2", "gamma")
+		requireNoResidue(t, jobs, "cat2", "delta")
+	})
+}

pkg/admin/catalog_handler.go

@@ -342,6 +342,10 @@ func (h *Handler) deleteCatalog(w http.ResponseWriter, r *http.Request) {
 	case err != nil:
 		writeError(w, http.StatusInternalServerError, "failed to delete catalog")
 	default:
+		// WithoutCancel: the delete has committed, so the queue cleanup
+		// must not die with the request (a client disconnect here would
+		// otherwise leave an open failure nothing else can resolve).
+		catalogindex.CancelCatalogBestEffort(context.WithoutCancel(r.Context()), h.deps.EmbedJobs, id)
 		writeJSON(w, http.StatusOK, statusResponse{Status: "deleted"})
 	}
 }
@@ -466,7 +470,7 @@ func (h *Handler) copyCatalogSpecs(w http.ResponseWriter, r *http.Request, srcID
 			// asynchronously. Without this the cloned spec
 			// would sit at "not indexed" until the periodic
 			// reconciler picked it up.
-			h.enqueueEmbedJob(r.Context(), dstID, s.SpecName)
+			catalogindex.EnqueueBestEffort(r.Context(), h.deps.EmbedJobs, dstID, s.SpecName)
 		}
 	}
 	return true
@@ -667,7 +671,7 @@ func (h *Handler) upsertCatalogSpec(w http.ResponseWriter, r *http.Request) {
 		writeError(w, h.specErrorStatus(err), "failed to save spec: "+err.Error())
 		return
 	}
-	h.enqueueEmbedJob(r.Context(), id, entry.SpecName)
+	catalogindex.EnqueueBestEffort(r.Context(), h.deps.EmbedJobs, id, entry.SpecName)
 	h.reloadConnectionsForCatalog(id)
 	saved, _ := h.deps.APICatalogStore.GetSpec(r.Context(), id, specName)
 	if saved != nil {
@@ -879,7 +883,7 @@ func (h *Handler) uploadCatalogSpec(w http.ResponseWriter, r *http.Request) {
 		writeError(w, h.specErrorStatus(err), "failed to save spec: "+err.Error())
 		return
 	}
-	h.enqueueEmbedJob(r.Context(), id, entry.SpecName)
+	catalogindex.EnqueueBestEffort(r.Context(), h.deps.EmbedJobs, id, entry.SpecName)
 	h.reloadConnectionsForCatalog(id)
 	saved, _ := h.deps.APICatalogStore.GetSpec(r.Context(), id, specName)
 	if saved != nil {
@@ -943,7 +947,7 @@ func (h *Handler) refreshCatalogSpec(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusInternalServerError, "failed to save refreshed spec: "+err.Error())
 		return
 	}
-	h.enqueueEmbedJob(r.Context(), id, entry.SpecName)
+	catalogindex.EnqueueBestEffort(r.Context(), h.deps.EmbedJobs, id, entry.SpecName)
 	h.reloadConnectionsForCatalog(id)
 	writeJSON(w, http.StatusOK, h.specToResponseWithEmbedding(r.Context(), id, entry, false))
 }
@@ -972,6 +976,9 @@ func (h *Handler) deleteCatalogSpec(w http.ResponseWriter, r *http.Request) {
 	case err != nil:
 		writeError(w, http.StatusInternalServerError, "failed to delete spec")
 	default:
+		// WithoutCancel: same rationale as deleteCatalog — the spec row
+		// is gone, so the cleanup outlives the request.
+		catalogindex.CancelBestEffort(context.WithoutCancel(r.Context()), h.deps.EmbedJobs, id, specName)
 		h.reloadConnectionsForCatalog(id)
 		writeJSON(w, http.StatusOK, statusResponse{Status: "deleted"})
 	}
@@ -1054,29 +1061,6 @@ func (h *Handler) specToResponseWithEmbedding(ctx context.Context, catalogID str
 	return resp
 }
 
-// enqueueEmbedJob is the producer-side hook every spec write
-// path calls after the spec row commits. It records the job
-// row alongside (or just after) the spec write and lets the
-// worker / reconciler / reaper drive the actual embedding pass
-// off the request path. Failures are logged but do not block
-// the spec write: the reconciler will pick up any spec whose
-// embedding-row count is below operation_count on its next
-// sweep, so a missed enqueue still converges.
-func (h *Handler) enqueueEmbedJob(ctx context.Context, catalogID, specName string) {
-	if h.deps.EmbedJobs == nil {
-		// No queue (file mode / no DB). The data path falls
-		// back to lexical and the operator gets no embeddings;
-		// this is the documented degraded mode.
-		return
-	}
-	if _, err := h.deps.EmbedJobs.Enqueue(ctx, catalogindex.SpecKey{
-		CatalogID: catalogID, SpecName: specName,
-	}, catalogindex.KindSpecWrite); err != nil {
-		slog.Warn("apigateway: enqueue embedding job failed",
-			logKeyCatalogID, logsan.SanitizeForLog(catalogID), logKeySpecName, logsan.SanitizeForLog(specName), logKeyError, err)
-	}
-}
-
 // logKeyCatalogID is the structured-log key for catalog ids in the
 // admin handler. Kept local to this file so other admin handlers
 // don't accidentally drift the spelling.