From 7c9e002ee11c393513a1d7c40e017d20459ee383 Mon Sep 17 00:00:00 2001 From: sun-970 <3843544764@qq.com> Date: Sun, 20 Sep 2026 09:18:34 +0000 Subject: [PATCH] fix(index): report unwired generation execution honestly (#174) Advertise execution_wired=false and fail-closed create/activate/rollback with 503 execution_unavailable so operators cannot enter stuck-building or empty-workspace active states. Do not implement a worker executor. --- CHANGELOG.md | 1 + docs/INDEX_GENERATIONS.md | 44 +++- .../api/handlers_index_generations.go | 40 ++-- .../api/handlers_index_generations_test.go | 190 +++++++++++++++++- 4 files changed, 239 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d4e0a13..aef4be6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ The project publishes 0.x prerelease versions; a stable release line is not yet ### Changed +- Index generation HTTP create/activate/rollback stay `503 execution_unavailable` with `execution_wired: false` on that body, and the same flag is now on events as well as list/status/cancel/resume/discard. Create still returns `400` for malformed JSON or an empty `profile_id` before the 503. Empty-workspace activate is refused by the same 503. Refs #174. - Ingest cursor locks try non-blocking exclusive locks and give up after 5s so a wedged peer becomes a warning instead of a silent hang. Refs #139. ## [0.1.2] - 2026-09-18 diff --git a/docs/INDEX_GENERATIONS.md b/docs/INDEX_GENERATIONS.md index d26da7b..664e531 100644 --- a/docs/INDEX_GENERATIONS.md +++ b/docs/INDEX_GENERATIONS.md @@ -105,23 +105,51 @@ content or raw provider responses. ## Current public surface -The HTTP and CLI surfaces are intentionally read-only in this foundation: +The HTTP and CLI surfaces expose read-only status and the full set of lifecycle +mutation routes. Create and activate return `execution_unavailable` (HTTP 503) +because no worker executor yet claims targets and search does not route +generation vectors. Cancel, resume, rollback and discard manage existing builds +but cannot produce a searchable corpus until execution is wired. ```text -GET /v1/workspaces/current/index-generations -GET /v1/workspaces/current/index-generations/{build-id} -GET /v1/workspaces/current/index-generations/{build-id}/events +GET /v1/workspaces/current/index-generations +GET /v1/workspaces/current/index-generations/{build-id} +GET /v1/workspaces/current/index-generations/{build-id}/events + +POST /v1/workspaces/current/index-generations → 503 execution_unavailable +POST /v1/workspaces/current/index-generations/{build-id}/cancel +POST /v1/workspaces/current/index-generations/{build-id}/resume +POST /v1/workspaces/current/index-generations/{build-id}/activate → 503 execution_unavailable +POST /v1/workspaces/current/index-generations/{build-id}/rollback → 503 execution_unavailable +POST /v1/workspaces/current/index-generations/{build-id}/discard mem generation list mem generation status mem generation events +mem generation create → rejected until execution is wired +mem generation activate → rejected until execution is wired +mem generation rollback → rejected until execution is wired +mem generation cancel +mem generation resume +mem generation discard ``` -They expose `execution_wired=false`. The server does not expose create, -activate, rollback, discard, cancel or resume yet. Publishing those mutations -before the Worker and search paths consume the same generation identity would -create a false state where metadata says “active” while queries still use the +Successful list, status, events, cancel, resume and discard responses include +`execution_wired: false`. Create, activate and rollback return +`503 execution_unavailable` with the same flag, and the HTTP handler never +calls `Service.Create`, `Service.Activate` or `Service.Rollback`. Those +service methods remain for in-process tests; they are not reachable over HTTP. +Create still validates JSON (`400 bad_json` / `bad_profile_id`) before the 503 +so a malformed body is not retried as a transient outage. +Cancel, resume and discard can mutate existing build metadata but cannot +produce a searchable corpus: the only HTTP writers of `active` state are +activate and rollback, and both are 503. +Empty-workspace activation is refused by the same 503: a generation with +`required_targets == 0` must not become `active` while search still uses the released legacy embedding tables. +Publishing a successful create or activate before the Worker and search paths +consume the same generation identity would create a false state where metadata +says “active” while queries still use the released legacy embedding tables. ## Cost, time and benchmark gate diff --git a/server/internal/api/handlers_index_generations.go b/server/internal/api/handlers_index_generations.go index 1857479..4145dce 100644 --- a/server/internal/api/handlers_index_generations.go +++ b/server/internal/api/handlers_index_generations.go @@ -37,7 +37,7 @@ func (s *Server) handleListIndexGenerations(w http.ResponseWriter, r *http.Reque } writeJSON(w, http.StatusOK, map[string]any{ "items": builds, - "execution_wired": true, + "execution_wired": false, }) } @@ -60,7 +60,7 @@ func (s *Server) handleGetIndexGeneration(w http.ResponseWriter, r *http.Request } writeJSON(w, http.StatusOK, map[string]any{ "generation": build, - "execution_wired": true, + "execution_wired": false, }) } @@ -81,7 +81,10 @@ func (s *Server) handleListIndexGenerationEvents(w http.ResponseWriter, r *http. writeIndexGenerationError(w, err) return } - writeJSON(w, http.StatusOK, map[string]any{"items": events}) + writeJSON(w, http.StatusOK, map[string]any{ + "items": events, + "execution_wired": false, + }) } func (s *Server) handleCreateIndexGeneration(w http.ResponseWriter, r *http.Request) { @@ -103,21 +106,11 @@ func (s *Server) handleCreateIndexGeneration(w http.ResponseWriter, r *http.Requ writeError(w, http.StatusBadRequest, "bad_json", err.Error()) return } - profileID := strings.TrimSpace(req.ProfileID) - if profileID == "" { + if strings.TrimSpace(req.ProfileID) == "" { writeError(w, http.StatusBadRequest, "bad_profile_id", "profile_id is required") return } - actor := r.Context().Value(ctxActor).(*auth.User) - build, err := s.IndexGenerations.Create(r.Context(), currentWorkspace(r).ID, actor.ID, profileID) - if err != nil { - writeIndexGenerationError(w, err) - return - } - writeJSON(w, http.StatusCreated, map[string]any{ - "generation": build, - "execution_wired": true, - }) + writeExecutionUnavailable(w, "index generation execution is not wired; no worker processes claimed targets") } func (s *Server) handleCancelIndexGeneration(w http.ResponseWriter, r *http.Request) { @@ -165,10 +158,9 @@ func (s *Server) indexGenerationBuildAction(w http.ResponseWriter, r *http.Reque build, err = s.IndexGenerations.Cancel(ctx, ws, actor.ID, id) case "resume": build, err = s.IndexGenerations.Resume(ctx, ws, actor.ID, id) - case "activate": - build, err = s.IndexGenerations.Activate(ctx, ws, actor.ID, id) - case "rollback": - build, err = s.IndexGenerations.Rollback(ctx, ws, actor.ID, id) + case "activate", "rollback": + writeExecutionUnavailable(w, "index generation execution is not wired; search does not route generation vectors") + return case "discard": build, err = s.IndexGenerations.Discard(ctx, ws, actor.ID, id) default: @@ -181,7 +173,7 @@ func (s *Server) indexGenerationBuildAction(w http.ResponseWriter, r *http.Reque } writeJSON(w, http.StatusOK, map[string]any{ "generation": build, - "execution_wired": true, + "execution_wired": false, }) } @@ -189,6 +181,14 @@ func indexGenerationBuildID(r *http.Request) (uuid.UUID, error) { return uuid.Parse(strings.TrimSpace(chi.URLParam(r, "buildID"))) } +func writeExecutionUnavailable(w http.ResponseWriter, hint string) { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "error": "execution_unavailable", + "hint": hint, + "execution_wired": false, + }) +} + func writeIndexGenerationError(w http.ResponseWriter, err error) { switch { case errors.Is(err, indexgeneration.ErrNotFound): diff --git a/server/internal/api/handlers_index_generations_test.go b/server/internal/api/handlers_index_generations_test.go index 35a389d..413005c 100644 --- a/server/internal/api/handlers_index_generations_test.go +++ b/server/internal/api/handlers_index_generations_test.go @@ -136,14 +136,17 @@ func TestIndexGenerationStatusHandlersStayWorkspaceScoped(t *testing.T) { } var response struct { Items []indexgeneration.Build `json:"items"` - ExecutionWired bool `json:"execution_wired"` + ExecutionWired *bool `json:"execution_wired"` } if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { t.Fatal(err) } - if len(response.Items) != 1 || response.Items[0].ID != buildID || !response.ExecutionWired { + if len(response.Items) != 1 || response.Items[0].ID != buildID { t.Fatalf("response = %#v", response) } + if response.ExecutionWired == nil || *response.ExecutionWired { + t.Fatalf("execution_wired = %v, want false", response.ExecutionWired) + } }) t.Run("get", func(t *testing.T) { @@ -195,14 +198,21 @@ func TestIndexGenerationMutationHandlers(t *testing.T) { request := indexGenerationMutationRequest(http.MethodPost, "/v1/workspaces/current/index-generations", workspaceID, actorID, "", body) server.handleCreateIndexGeneration(recorder, request) - if recorder.Code != http.StatusCreated { + if recorder.Code != http.StatusServiceUnavailable { t.Fatalf("status = %d; body = %s", recorder.Code, recorder.Body.String()) } - if service.lastProfile != "local-fast-v2" { - t.Fatalf("profile = %q", service.lastProfile) + var response map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response["error"] != "execution_unavailable" { + t.Fatalf("error = %v, want execution_unavailable", response["error"]) + } + if response["execution_wired"] != false { + t.Fatalf("execution_wired = %v, want false", response["execution_wired"]) } - if service.lastActor != actorID { - t.Fatalf("actor = %s, want %s", service.lastActor, actorID) + if service.lastProfile != "" { + t.Fatalf("service.Create should not have been called, but profile = %q", service.lastProfile) } }) @@ -217,6 +227,17 @@ func TestIndexGenerationMutationHandlers(t *testing.T) { } }) + t.Run("create_rejects_malformed_body", func(t *testing.T) { + recorder := httptest.NewRecorder() + body := strings.NewReader(`{{{{`) + request := indexGenerationMutationRequest(http.MethodPost, + "/v1/workspaces/current/index-generations", workspaceID, actorID, "", body) + server.handleCreateIndexGeneration(recorder, request) + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status = %d; body = %s", recorder.Code, recorder.Body.String()) + } + }) + t.Run("cancel", func(t *testing.T) { recorder := httptest.NewRecorder() request := indexGenerationMutationRequest(http.MethodPost, @@ -254,9 +275,162 @@ func TestIndexGenerationMutationHandlers(t *testing.T) { t.Fatalf("status = %d; body = %s", recorder.Code, recorder.Body.String()) } }) + + t.Run("activate_blocked", func(t *testing.T) { + service.lastAction = "" + recorder := httptest.NewRecorder() + request := indexGenerationMutationRequest(http.MethodPost, + "/v1/workspaces/current/index-generations/"+buildID.String()+"/activate", + workspaceID, actorID, buildID.String(), nil) + server.handleActivateIndexGeneration(recorder, request) + if recorder.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d; body = %s", recorder.Code, recorder.Body.String()) + } + var response map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response["error"] != "execution_unavailable" { + t.Fatalf("error = %v, want execution_unavailable", response["error"]) + } + if response["execution_wired"] != false { + t.Fatalf("execution_wired = %v, want false", response["execution_wired"]) + } + if service.lastAction != "" { + t.Fatalf("service.Activate should not have been called, but action = %q", service.lastAction) + } + }) + + t.Run("rollback_blocked", func(t *testing.T) { + service.lastAction = "" + recorder := httptest.NewRecorder() + request := indexGenerationMutationRequest(http.MethodPost, + "/v1/workspaces/current/index-generations/"+buildID.String()+"/rollback", + workspaceID, actorID, buildID.String(), nil) + server.handleRollbackIndexGeneration(recorder, request) + if recorder.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d; body = %s", recorder.Code, recorder.Body.String()) + } + var response map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response["error"] != "execution_unavailable" { + t.Fatalf("error = %v, want execution_unavailable", response["error"]) + } + if response["execution_wired"] != false { + t.Fatalf("execution_wired = %v, want false", response["execution_wired"]) + } + if service.lastAction != "" { + t.Fatalf("service.Rollback should not have been called, but action = %q", service.lastAction) + } + }) +} + +func TestExecutionWiredFlagMatchesCapability(t *testing.T) { + workspaceID := uuid.New() + actorID := uuid.New() + buildID := uuid.New() + service := &fakeIndexGenerationService{buildID: buildID} + server := &Server{IndexGenerations: service} + + assertWiredFalse := func(t *testing.T, body []byte) { + t.Helper() + var payload map[string]any + if err := json.Unmarshal(body, &payload); err != nil { + t.Fatal(err) + } + wired, ok := payload["execution_wired"] + if !ok { + t.Fatal("execution_wired key is missing") + } + if wired != false { + t.Fatalf("execution_wired = %#v, want false", wired) + } + } + + listRecorder := httptest.NewRecorder() + listRequest := indexGenerationRequest(http.MethodGet, + "/v1/workspaces/current/index-generations?limit=25", workspaceID, "") + server.handleListIndexGenerations(listRecorder, listRequest) + if listRecorder.Code != http.StatusOK { + t.Fatalf("list status = %d; body = %s", listRecorder.Code, listRecorder.Body.String()) + } + assertWiredFalse(t, listRecorder.Body.Bytes()) + + eventsRecorder := httptest.NewRecorder() + eventsRequest := indexGenerationRequest(http.MethodGet, + "/v1/workspaces/current/index-generations/"+buildID.String()+"/events", workspaceID, buildID.String()) + server.handleListIndexGenerationEvents(eventsRecorder, eventsRequest) + if eventsRecorder.Code != http.StatusOK { + t.Fatalf("events status = %d; body = %s", eventsRecorder.Code, eventsRecorder.Body.String()) + } + assertWiredFalse(t, eventsRecorder.Body.Bytes()) + + createRecorder := httptest.NewRecorder() + createBody := strings.NewReader(`{"profile_id":"local-fast-v2"}`) + createRequest := indexGenerationMutationRequest(http.MethodPost, + "/v1/workspaces/current/index-generations", workspaceID, actorID, "", createBody) + server.handleCreateIndexGeneration(createRecorder, createRequest) + if createRecorder.Code == http.StatusCreated { + t.Fatal("create must not succeed when execution_wired is false") + } + if createRecorder.Code != http.StatusServiceUnavailable { + t.Fatalf("create status = %d; body = %s", createRecorder.Code, createRecorder.Body.String()) + } + assertWiredFalse(t, createRecorder.Body.Bytes()) + if service.lastProfile != "" { + t.Fatalf("create must not call Service.Create, profile = %q", service.lastProfile) + } + + activateRecorder := httptest.NewRecorder() + activateRequest := indexGenerationMutationRequest(http.MethodPost, + "/v1/workspaces/current/index-generations/"+buildID.String()+"/activate", + workspaceID, actorID, buildID.String(), nil) + server.handleActivateIndexGeneration(activateRecorder, activateRequest) + if activateRecorder.Code == http.StatusOK { + t.Fatal("activate must not succeed when execution_wired is false") + } + if activateRecorder.Code != http.StatusServiceUnavailable { + t.Fatalf("activate status = %d; body = %s", activateRecorder.Code, activateRecorder.Body.String()) + } + assertWiredFalse(t, activateRecorder.Body.Bytes()) + if service.lastAction != "" { + t.Fatalf("activate must not call Service.Activate, action = %q", service.lastAction) + } +} + +func TestEmptyWorkspaceActivateIsRefused(t *testing.T) { + // Empty-workspace activation would otherwise pass the 0==0 quality gate + // and mark a generation with no vectors active. Activate is refused for + // every workspace while execution is unwired, including the empty case. + workspaceID := uuid.New() + actorID := uuid.New() + buildID := uuid.New() + service := &fakeIndexGenerationService{buildID: buildID} + server := &Server{IndexGenerations: service} + + recorder := httptest.NewRecorder() + request := indexGenerationMutationRequest(http.MethodPost, + "/v1/workspaces/current/index-generations/"+buildID.String()+"/activate", + workspaceID, actorID, buildID.String(), nil) + server.handleActivateIndexGeneration(recorder, request) + if recorder.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d; body = %s", recorder.Code, recorder.Body.String()) + } + var response map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response["error"] != "execution_unavailable" { + t.Fatalf("error = %v", response["error"]) + } + if service.lastAction != "" { + t.Fatalf("empty-workspace activate must not call Service.Activate") + } } -func TestIndexGenerationPublicRoutesAreReadOnly(t *testing.T) { +func TestIndexGenerationPublicRoutesAreRegistered(t *testing.T) { routes, ok := (&Server{}).Router().(chi.Routes) if !ok { t.Fatal("Server.Router did not return chi.Routes")