Idempotent ingest + self-diagnosing not-found (HAL-323) (#49)

* engine: idempotent ingest + self-diagnosing not-found (HAL-323)

Two-part fix for flaky bulk ingestion and the bench's duplicate uploads.

Idempotent ingest (root cause of "same doc ingested up to 6x"):
- New `idempotency_key` column on documents + partial unique index on
  (org_id, idempotency_key) — migration 0007. NULL key = no dedup
  (preserves current behavior for callers that don't opt in).
- All three ingest handlers (standalone/local in internal/api, multi-tenant
  REST in internal/handler, Connect RPC in internal/connecthandler) honor the
  `Idempotency-Key` header the SDK already sends: a repeat ingest returns the
  original document instead of creating a duplicate, and the concurrent
  same-key race is resolved by dropping the orphan source and returning the
  winner. The SDK retry-on-transient (which re-POSTs after a reset that landed
  AFTER the server committed) is now a no-op.
- db: ErrConflict sentinel mapped from SQLSTATE 23505; GetDocumentByIdempotencyKey.

Self-diagnosing not-found (resolves the pathless "object not found" mystery):
- Every storage backend's ErrNotFound now carries the resolved key/path
  (local Get+Delete, s3 Get+Delete, gcs Get+Delete) so a failure can always be
  attributed to a code path — pathless rows were pre-path-wrap leftovers.
- local.Get: the open-after-stat result is wrapped as retryable ErrNotFound
  (previously a raw error getSourceWithRetry would NOT retry), and a Windows-only
  bounded retry rides through the Defender scan window that transiently hides a
  freshly-written file.

Docs: OpenAPI documents the Idempotency-Key header; README documents the
mandatory Windows Defender exclusion for the storage root.

Verified: go build/vet clean; storage + db (live-Postgres) tests pass; and an
end-to-end double-POST with one Idempotency-Key returns one document_id while a
keyless control gets its own.

* engine: gofmt — align idempotency_key struct field

* engine: address PR review — scope idempotency replay by org, log orphan cleanup

- middleware.Idempotency: compose the cache key as method|path|org|key instead
  of the bare Idempotency-Key, so two tenants reusing the same opaque key can't
  receive each other's cached 202 (cross-tenant leak). New tests cover same-org
  replay, cross-org isolation, and keyless pass-through. (CodeRabbit, Major)
- All three ingest handlers now log orphan-source cleanup failures on the
  idempotency-conflict path instead of discarding the Delete error. (Sourcery +
  CodeRabbit)
- GetDocumentByIdempotencyKey hydrates IdempotencyKey from the DB column.
- Strengthen TestLocalGetRetriesTransientNotExist: on Windows it now asserts the
  FIRST Get succeeds, so the test fails if the retry loop is removed. (Sourcery)
- README: "freshly written" typo.

Author: hallelx2

README.md

@@ -164,6 +164,15 @@ llm:
 
 > **Anthropic-compatible gateways (GLM/Z.ai):** `base_url` **must include `/v1`** — the client posts to `${base_url}/messages`.
 
+> **Windows + local storage:** Windows Defender real-time protection scans
+> each freshly written file and briefly hides it from `os.Stat`/`os.Open`,
+> which under heavy concurrent ingestion can surface as transient
+> `object not found` errors. The local backend rides through this window with
+> a short internal retry, but for large bulk loads **add a Defender exclusion
+> for your storage root** (`local.root` / `VLE_STORAGE_LOCAL_ROOT`):
+> `Add-MpPreference -ExclusionPath "C:\path\to\data\documents"`. Linux has no
+> such scan-hold and needs no exclusion.
+
 ### Supported formats
 
 PDF (positioned text + tables via [`pdftable`](https://github.com/hallelx2/pdftable)) · Markdown · HTML · DOCX · Text.

internal/api/server.go

@@ -231,6 +231,28 @@ func (d Deps) handleIngestDocument(w http.ResponseWriter, r *http.Request) {
 
 	docID := ingest.NewDocumentID()
 
+	// Idempotent ingest: if the caller supplied an Idempotency-Key and a
+	// document already exists under it (for the standalone org), return that
+	// document instead of creating a duplicate. A client — or the SDK's own
+	// transport retry-on-transient — that re-POSTs the same upload after a
+	// reset that landed AFTER the server committed the row is then a no-op.
+	// This is the standalone/local-mode half of the HAL-323 duplicate-upload
+	// fix (the multi-tenant handler in internal/handler mirrors it).
+	idemKey := r.Header.Get("Idempotency-Key")
+	if idemKey != "" {
+		if existing, err := d.DB.GetDocumentByIdempotencyKey(ctx, standaloneOrgID, idemKey); err == nil {
+			writeJSON(w, http.StatusAccepted, map[string]any{
+				"document_id": existing.ID,
+				"status":      string(existing.Status),
+			})
+			return
+		} else if !errors.Is(err, db.ErrNotFound) {
+			d.Logger.Error("ingest: idempotency lookup failed", "err", err)
+			writeErr(w, http.StatusInternalServerError, "db read failed")
+			return
+		}
+	}
+
 	var (
 		filename    string
 		contentType string
@@ -300,14 +322,29 @@ func (d Deps) handleIngestDocument(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if err := d.DB.NewDocument(ctx, db.Document{
-		ID:          docID,
-		OrgID:       standaloneOrgID,
-		Title:       title,
-		ContentType: contentType,
-		SourceRef:   key,
-		Status:      db.StatusPending,
-		ByteSize:    size,
+		ID:             docID,
+		OrgID:          standaloneOrgID,
+		Title:          title,
+		ContentType:    contentType,
+		SourceRef:      key,
+		Status:         db.StatusPending,
+		ByteSize:       size,
+		IdempotencyKey: idemKey,
 	}); err != nil {
+		// A concurrent same-key ingest won the race: drop the orphan source we
+		// just wrote and return the winner instead of a 500.
+		if idemKey != "" && errors.Is(err, db.ErrConflict) {
+			if delErr := d.Storage.Delete(ctx, key); delErr != nil {
+				d.Logger.Warn("ingest: orphan source cleanup failed", "err", delErr, "source_ref", key, "idempotency_key", idemKey)
+			}
+			if existing, lookupErr := d.DB.GetDocumentByIdempotencyKey(ctx, standaloneOrgID, idemKey); lookupErr == nil {
+				writeJSON(w, http.StatusAccepted, map[string]any{
+					"document_id": existing.ID,
+					"status":      string(existing.Status),
+				})
+				return
+			}
+		}
 		d.Logger.Error("ingest: db insert failed", "err", err)
 		writeErr(w, http.StatusInternalServerError, "db write failed")
 		return

internal/connecthandler/documents.go

@@ -82,6 +82,23 @@ func (s *DocumentsService) CreateDocument(
 		return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("content is required"))
 	}
 
+	// Idempotent ingest: a repeat CreateDocument carrying the same
+	// Idempotency-Key returns the original document rather than a duplicate
+	// (mirrors the REST path; closes the HAL-323 duplicate-upload bug for SDK
+	// callers that retry the RPC).
+	idemKey := req.Header().Get("Idempotency-Key")
+	if idemKey != "" {
+		if existing, err := s.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); err == nil {
+			return connect.NewResponse(&v1.CreateDocumentResponse{
+				DocumentId: string(existing.ID),
+				Status:     string(existing.Status),
+			}), nil
+		} else if !errors.Is(err, db.ErrNotFound) {
+			s.logger.Error("ingest: idempotency lookup failed", "err", err)
+			return nil, connect.NewError(connect.CodeInternal, errors.New("db read failed"))
+		}
+	}
+
 	docID := ingest.NewDocumentID()
 	contentType := msg.ContentType
 	if contentType == "" {
@@ -105,15 +122,29 @@ func (s *DocumentsService) CreateDocument(
 	}
 
 	if err := s.db.NewDocument(ctx, db.Document{
-		ID:          docID,
-		OrgID:       orgID,
-		StoreID:     storeIDFromConnect(req),
-		Title:       title,
-		ContentType: contentType,
-		SourceRef:   key,
-		Status:      db.StatusPending,
-		ByteSize:    size,
+		ID:             docID,
+		OrgID:          orgID,
+		StoreID:        storeIDFromConnect(req),
+		Title:          title,
+		ContentType:    contentType,
+		SourceRef:      key,
+		Status:         db.StatusPending,
+		ByteSize:       size,
+		IdempotencyKey: idemKey,
 	}); err != nil {
+		// Concurrent same-key insert won the race: drop the orphan source and
+		// return the winner instead of erroring.
+		if idemKey != "" && errors.Is(err, db.ErrConflict) {
+			if delErr := s.storage.Delete(ctx, key); delErr != nil {
+				s.logger.Warn("ingest: orphan source cleanup failed", "err", delErr, "source_ref", key, "idempotency_key", idemKey)
+			}
+			if existing, lookupErr := s.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); lookupErr == nil {
+				return connect.NewResponse(&v1.CreateDocumentResponse{
+					DocumentId: string(existing.ID),
+					Status:     string(existing.Status),
+				}), nil
+			}
+		}
 		s.logger.Error("ingest: db insert failed", "err", err)
 		return nil, connect.NewError(connect.CodeInternal, errors.New("db write failed"))
 	}

internal/handler/documents.go

@@ -158,6 +158,25 @@ func (h *DocumentsHandler) HandleIngestDocument(w http.ResponseWriter, r *http.R
 	ctx := r.Context()
 	docID := ingest.NewDocumentID()
 
+	// Idempotent ingest: if the client supplied an Idempotency-Key and we
+	// already have a document for this org under that key, return it instead
+	// of creating a duplicate. This makes a client/transport retry of the
+	// ingest POST (the HAL-323 duplicate-upload bug) a no-op.
+	idemKey := r.Header.Get("Idempotency-Key")
+	if idemKey != "" {
+		if existing, err := h.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); err == nil {
+			writeJSON(w, http.StatusAccepted, map[string]any{
+				"document_id": existing.ID,
+				"status":      string(existing.Status),
+			})
+			return
+		} else if !errors.Is(err, db.ErrNotFound) {
+			h.logger.Error("ingest: idempotency lookup failed", "err", err)
+			writeErr(w, http.StatusInternalServerError, "db read failed")
+			return
+		}
+	}
+
 	var (
 		filename    string
 		contentType string
@@ -227,15 +246,31 @@ func (h *DocumentsHandler) HandleIngestDocument(w http.ResponseWriter, r *http.R
 	}
 
 	if err := h.db.NewDocument(ctx, db.Document{
-		ID:          docID,
-		OrgID:       orgID,
-		StoreID:     storeID(r),
-		Title:       title,
-		ContentType: contentType,
-		SourceRef:   key,
-		Status:      db.StatusPending,
-		ByteSize:    size,
+		ID:             docID,
+		OrgID:          orgID,
+		StoreID:        storeID(r),
+		Title:          title,
+		ContentType:    contentType,
+		SourceRef:      key,
+		Status:         db.StatusPending,
+		ByteSize:       size,
+		IdempotencyKey: idemKey,
 	}); err != nil {
+		// A concurrent request with the same Idempotency-Key won the race and
+		// inserted first. Drop the orphan source we just wrote and return the
+		// winner's document rather than a 500.
+		if idemKey != "" && errors.Is(err, db.ErrConflict) {
+			if delErr := h.storage.Delete(ctx, key); delErr != nil {
+				h.logger.Warn("ingest: orphan source cleanup failed", "err", delErr, "source_ref", key, "idempotency_key", idemKey)
+			}
+			if existing, lookupErr := h.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); lookupErr == nil {
+				writeJSON(w, http.StatusAccepted, map[string]any{
+					"document_id": existing.ID,
+					"status":      string(existing.Status),
+				})
+				return
+			}
+		}
 		h.logger.Error("ingest: db insert failed", "err", err)
 		writeErr(w, http.StatusInternalServerError, "db write failed")
 		return

internal/middleware/idempotency.go

@@ -122,13 +122,22 @@ func Idempotency(cfg IdempotencyConfig) func(http.Handler) http.Handler {
 				return
 			}
 
-			key := r.Header.Get("Idempotency-Key")
-			if key == "" {
+			rawKey := r.Header.Get("Idempotency-Key")
+			if rawKey == "" {
 				// No key provided — proceed normally.
 				next.ServeHTTP(w, r)
 				return
 			}
 
+			// Scope the cache entry by tenant + method + path, NOT the raw
+			// Idempotency-Key alone. Two orgs legitimately use the same opaque
+			// key (e.g. "1", a per-corpus id); replaying by the bare key would
+			// hand org B a 202 cached for org A's document — a cross-tenant
+			// leak. The org header is absent in standalone/local mode, which is
+			// single-tenant, so an empty org collapses to one namespace there.
+			org := r.Header.Get("X-Vectorless-Org")
+			key := r.Method + "|" + r.URL.Path + "|" + org + "|" + rawKey
+
 			// Check cache.
 			if cr, ok := cache.get(key); ok {
 				w.Header().Set("X-Idempotency-Replayed", "true")

internal/middleware/idempotency_test.go

@@ -0,0 +1,89 @@
+package middleware
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+)
+
+// handler that returns a per-org body and counts invocations, so we can tell a
+// cached replay from a real pass-through.
+func countingOrgHandler(calls *atomic.Int64) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls.Add(1)
+		w.WriteHeader(http.StatusAccepted)
+		_, _ = w.Write([]byte("doc-for-" + r.Header.Get("X-Vectorless-Org")))
+	})
+}
+
+func post(mw http.Handler, org, key string) *httptest.ResponseRecorder {
+	req := httptest.NewRequest(http.MethodPost, "/v1/documents", nil)
+	if org != "" {
+		req.Header.Set("X-Vectorless-Org", org)
+	}
+	if key != "" {
+		req.Header.Set("Idempotency-Key", key)
+	}
+	rec := httptest.NewRecorder()
+	mw.ServeHTTP(rec, req)
+	return rec
+}
+
+func body(rec *httptest.ResponseRecorder) string {
+	return rec.Body.String()
+}
+
+func TestIdempotencySameOrgReplays(t *testing.T) {
+	var calls atomic.Int64
+	mw := Idempotency(IdempotencyConfig{})(countingOrgHandler(&calls))
+
+	first := post(mw, "orgA", "k1")
+	second := post(mw, "orgA", "k1")
+
+	if calls.Load() != 1 {
+		t.Fatalf("handler invoked %d times, want 1 (second should replay)", calls.Load())
+	}
+	if got := second.Header().Get("X-Idempotency-Replayed"); got != "true" {
+		t.Fatalf("second response missing replay header, got %q", got)
+	}
+	if body(first) != body(second) {
+		t.Fatalf("replay body %q != original %q", body(second), body(first))
+	}
+}
+
+func TestIdempotencyDifferentOrgsDoNotCollide(t *testing.T) {
+	// The core cross-tenant guarantee: two orgs sending the SAME opaque
+	// Idempotency-Key must each hit the handler and get THEIR OWN response —
+	// org B must never be handed org A's cached document.
+	var calls atomic.Int64
+	mw := Idempotency(IdempotencyConfig{})(countingOrgHandler(&calls))
+
+	a := post(mw, "orgA", "shared-key")
+	b := post(mw, "orgB", "shared-key")
+
+	if calls.Load() != 2 {
+		t.Fatalf("handler invoked %d times, want 2 (no cross-org collision)", calls.Load())
+	}
+	if b.Header().Get("X-Idempotency-Replayed") == "true" {
+		t.Fatal("org B got a replayed response — cross-tenant leak")
+	}
+	if body(a) == body(b) {
+		t.Fatalf("org B received org A's body %q — cross-tenant leak", body(b))
+	}
+	if body(b) != "doc-for-orgB" {
+		t.Fatalf("org B body = %q, want doc-for-orgB", body(b))
+	}
+}
+
+func TestIdempotencyNoKeyAlwaysPassesThrough(t *testing.T) {
+	var calls atomic.Int64
+	mw := Idempotency(IdempotencyConfig{})(countingOrgHandler(&calls))
+
+	post(mw, "orgA", "")
+	post(mw, "orgA", "")
+
+	if calls.Load() != 2 {
+		t.Fatalf("handler invoked %d times, want 2 (no key = no caching)", calls.Load())
+	}
+}

openapi.yaml

@@ -113,6 +113,21 @@ paths:
         (field: "file") or JSON body with content. Returns immediately
         with a document_id in "pending" state. The ingest pipeline runs
         asynchronously: parse → build tree → summarize sections.
+
+        Pass an `Idempotency-Key` header to make ingestion idempotent: a
+        repeated request with the same key returns the original document
+        (and its current status) instead of creating a duplicate. Use this
+        to make a client/network retry of the upload safe.
+      parameters:
+        - name: Idempotency-Key
+          in: header
+          required: false
+          description: |
+            Opaque client-chosen key, unique per document. Repeating a
+            request with the same key returns the existing document rather
+            than ingesting a duplicate.
+          schema:
+            type: string
       requestBody:
         required: true
         content:

pkg/db/db.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
@@ -25,6 +26,10 @@ var migrationsFS embed.FS
 // ErrNotFound signals a missing row.
 var ErrNotFound = errors.New("db: not found")
 
+// ErrConflict signals a unique-constraint violation (SQLSTATE 23505), e.g.
+// a second insert with the same (org_id, idempotency_key).
+var ErrConflict = errors.New("db: conflict")
+
 // Pool wraps *pgxpool.Pool with engine-specific helpers.
 type Pool struct {
 	*pgxpool.Pool
@@ -136,5 +141,9 @@ func mapErr(err error) error {
 	if errors.Is(err, pgx.ErrNoRows) {
 		return ErrNotFound
 	}
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+		return fmt.Errorf("%w: %s", ErrConflict, pgErr.ConstraintName)
+	}
 	return err
 }