From 1fa352cf28fa6fa446109a95b6fd7f7c31ce61c4 Mon Sep 17 00:00:00 2001 From: Obayo Tian Date: Sat, 18 Jul 2026 14:51:30 +0800 Subject: [PATCH] fix: enforce parent child issue state constraint --- server/internal/handler/github.go | 10 + server/internal/handler/issue.go | 177 ++++++- .../internal/handler/issue_child_done_test.go | 47 +- .../issue_parent_state_constraint_test.go | 474 ++++++++++++++++++ server/internal/issueguard/parent_state.go | 62 +++ server/internal/service/task.go | 18 +- ...202_issue_parent_state_constraint.down.sql | 2 + .../202_issue_parent_state_constraint.up.sql | 143 ++++++ 8 files changed, 889 insertions(+), 44 deletions(-) create mode 100644 server/internal/handler/issue_parent_state_constraint_test.go create mode 100644 server/internal/issueguard/parent_state.go create mode 100644 server/migrations/202_issue_parent_state_constraint.down.sql create mode 100644 server/migrations/202_issue_parent_state_constraint.up.sql diff --git a/server/internal/handler/github.go b/server/internal/handler/github.go index 1f45430af77..91c0a3fa96d 100644 --- a/server/internal/handler/github.go +++ b/server/internal/handler/github.go @@ -23,6 +23,7 @@ import ( "github.com/golang-jwt/jwt/v5" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" + "github.com/multica-ai/multica/server/internal/issueguard" "github.com/multica-ai/multica/server/internal/middleware" db "github.com/multica-ai/multica/server/pkg/db/generated" "github.com/multica-ai/multica/server/pkg/protocol" @@ -1367,6 +1368,15 @@ func (h *Handler) advanceIssueToDone(ctx context.Context, issue db.Issue, worksp WorkspaceID: issue.WorkspaceID, }) if err != nil { + if conflict, ok := issueguard.ParentStateConflictFrom(err); ok { + slog.Warn("github: automatic issue completion rejected by parent-state constraint", + "issue_id", issue.ID, + "workspace_id", issue.WorkspaceID, + "parent_issue_id", conflict.ParentIssueID, + "conflict_code", conflict.Code, + ) + return + } slog.Warn("github: advance issue to done failed", "err", err) return } diff --git a/server/internal/handler/issue.go b/server/internal/handler/issue.go index 82ca84338fc..a0dcc54d6bc 100644 --- a/server/internal/handler/issue.go +++ b/server/internal/handler/issue.go @@ -10,6 +10,7 @@ import ( "net/http" "net/url" "regexp" + "sort" "strconv" "strings" "time" @@ -2381,6 +2382,27 @@ func duplicateIssueMessage(issue IssueResponse) string { return issueguard.DuplicateMessage(issue.Identifier, issue.Title, issue.Status) } +// writeParentStateConflict translates the database-owned hierarchy invariant +// into the stable public API contract. The database remains the single source +// of truth for HTTP, workers, integrations, and imports; this helper only +// shapes its safe, aggregate error for HTTP clients. +func writeParentStateConflict(w http.ResponseWriter, err error) bool { + conflict, ok := issueguard.ParentStateConflictFrom(err) + if !ok { + return false + } + payload := map[string]any{ + "code": conflict.Code, + "error": issueguard.ParentStateConflictMessage(conflict.Code), + "parent_issue_id": conflict.ParentIssueID, + } + if conflict.IncompleteDescendantCount != nil { + payload["incomplete_descendant_count"] = *conflict.IncompleteDescendantCount + } + writeJSON(w, http.StatusConflict, payload) + return true +} + func (h *Handler) CreateIssue(w http.ResponseWriter, r *http.Request) { var req CreateIssueRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -2620,6 +2642,9 @@ func (h *Handler) CreateIssue(w http.ResponseWriter, r *http.Request) { }) return } + if writeParentStateConflict(w, err) { + return + } if errors.Is(err, service.ErrParentIssueNotFound) { writeError(w, http.StatusBadRequest, "parent issue not found in this workspace") return @@ -2866,6 +2891,9 @@ func (h *Handler) UpdateIssue(w http.ResponseWriter, r *http.Request) { issue, err := h.Queries.UpdateIssue(r.Context(), params) if err != nil { + if writeParentStateConflict(w, err) { + return + } slog.Warn("update issue failed", append(logger.RequestAttrs(r), "error", err, "issue_id", id, "workspace_id", workspaceID)...) writeError(w, http.StatusInternalServerError, "failed to update issue: "+err.Error()) return @@ -3179,6 +3207,91 @@ type BatchUpdateIssuesRequest struct { Updates UpdateIssueRequest `json:"updates"` } +type batchUpdatedIssue struct { + issue db.Issue + prevIssue db.Issue +} + +// orderBatchIssueIDsForStatus makes a uniform status update independent of +// request order for issues that belong to the same selected tree. Terminal +// transitions must visit descendants first; active transitions must visit +// ancestors first. The database trigger remains the authority for concurrent +// hierarchy changes, so an unavailable or out-of-selection ancestor simply +// keeps its stable request order here. +func (h *Handler) orderBatchIssueIDsForStatus(ctx context.Context, workspaceID pgtype.UUID, issueIDs []string, terminal bool) []string { + type orderedIssue struct { + issueID string + canonicalID string + depth int + known bool + } + + ordered := make([]orderedIssue, len(issueIDs)) + issuesByID := make(map[string]db.Issue, len(issueIDs)) + for index, issueID := range issueIDs { + ordered[index].issueID = issueID + issueUUID, err := util.ParseUUID(issueID) + if err != nil { + continue + } + issue, err := h.Queries.GetIssueInWorkspace(ctx, db.GetIssueInWorkspaceParams{ + ID: issueUUID, + WorkspaceID: workspaceID, + }) + if err != nil { + continue + } + ordered[index].canonicalID = uuidToString(issue.ID) + issuesByID[ordered[index].canonicalID] = issue + ordered[index].known = true + } + + var depthFor func(string, map[string]bool) int + depthFor = func(issueID string, path map[string]bool) int { + issue, ok := issuesByID[issueID] + if !ok || !issue.ParentIssueID.Valid { + return 0 + } + parentID := uuidToString(issue.ParentIssueID) + if path[parentID] { + return 0 + } + if _, ok := issuesByID[parentID]; !ok { + return 0 + } + path[parentID] = true + defer delete(path, parentID) + return 1 + depthFor(parentID, path) + } + for index := range ordered { + if ordered[index].known { + ordered[index].depth = depthFor(ordered[index].canonicalID, map[string]bool{ordered[index].canonicalID: true}) + } + } + + sort.SliceStable(ordered, func(left, right int) bool { + if ordered[left].known != ordered[right].known { + // Unknown or stale selections are skipped by the existing batch + // contract. Keep them after known rows so they cannot break the + // strict tree ordering of a valid parent/child selection. + return ordered[left].known + } + if !ordered[left].known { + return false + } + if terminal { + return ordered[left].depth > ordered[right].depth + } + return ordered[left].depth < ordered[right].depth + }) + + result := make([]string, len(ordered)) + for index, issue := range ordered { + result[index] = issue.issueID + } + return result +} + func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { bodyBytes, err := io.ReadAll(r.Body) if err != nil { @@ -3251,17 +3364,34 @@ func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { if !ok { return } + orderedIssueIDs := req.IssueIDs + if req.Updates.Status != nil { + orderedIssueIDs = h.orderBatchIssueIDsForStatus(r.Context(), wsUUID, req.IssueIDs, isTerminalChildStatus(*req.Updates.Status)) + } + + // A parent-state conflict must reject the whole batch. Keep every row write + // in one transaction and defer event/automation side effects until after the + // commit, so a 409 cannot leave earlier batch entries visible. + tx, err := h.TxStarter.Begin(r.Context()) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to start batch transaction") + return + } + defer tx.Rollback(r.Context()) + qtx := h.Queries.WithTx(tx) + updated := 0 + updatedIssues := make([]batchUpdatedIssue, 0, len(orderedIssueIDs)) // Children that transitioned into a terminal status this batch, collected so // the parent/stage notification is evaluated once against the final state // after the loop (MUL-4155) rather than per-child mid-batch. var childDoneCompleted []db.Issue - for _, issueID := range req.IssueIDs { + for _, issueID := range orderedIssueIDs { issueUUID, err := util.ParseUUID(issueID) if err != nil { continue } - prevIssue, err := h.Queries.GetIssueInWorkspace(r.Context(), db.GetIssueInWorkspaceParams{ + prevIssue, err := qtx.GetIssueInWorkspace(r.Context(), db.GetIssueInWorkspaceParams{ ID: issueUUID, WorkspaceID: wsUUID, }) @@ -3347,7 +3477,7 @@ func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { continue } // Validate parent exists in the same workspace. - if _, err := h.Queries.GetIssueInWorkspace(r.Context(), db.GetIssueInWorkspaceParams{ + if _, err := qtx.GetIssueInWorkspace(r.Context(), db.GetIssueInWorkspaceParams{ ID: newParentID, WorkspaceID: prevIssue.WorkspaceID, }); err != nil { @@ -3357,7 +3487,7 @@ func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { cycleDetected := false cursor := newParentID for depth := 0; depth < 10; depth++ { - ancestor, err := h.Queries.GetIssue(r.Context(), cursor) + ancestor, err := qtx.GetIssue(r.Context(), cursor) if err != nil || !ancestor.ParentIssueID.Valid { break } @@ -3407,16 +3537,30 @@ func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { } } - issue, err := h.Queries.UpdateIssue(r.Context(), params) + issue, err := qtx.UpdateIssue(r.Context(), params) if err != nil { + if writeParentStateConflict(w, err) { + return + } slog.Warn("batch update issue failed", "issue_id", issueID, "error", err) - continue + writeError(w, http.StatusInternalServerError, "failed to update issue in batch") + return } + updatedIssues = append(updatedIssues, batchUpdatedIssue{issue: issue, prevIssue: prevIssue}) + updated++ + } + if err := tx.Commit(r.Context()); err != nil { + writeError(w, http.StatusInternalServerError, "failed to commit batch update") + return + } + + for _, updatedIssue := range updatedIssues { + issue := updatedIssue.issue + prevIssue := updatedIssue.prevIssue prefix := h.getIssuePrefix(r.Context(), issue.WorkspaceID) resp := issueToResponse(issue, prefix) actorType, actorID := h.resolveActor(r, userID, workspaceID) - assigneeChanged := (req.Updates.AssigneeType != nil || req.Updates.AssigneeID != nil) && (prevIssue.AssigneeType.String != issue.AssigneeType.String || uuidToString(prevIssue.AssigneeID) != uuidToString(issue.AssigneeID)) statusChanged := req.Updates.Status != nil && prevIssue.Status != issue.Status @@ -3431,12 +3575,6 @@ func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { "project_changed": projectChanged, }) - // Reassignment does not cancel existing tasks (#4963 / MUL-4113) — - // mirrors UpdateIssue. See that handler for the rationale. - // - // Same single predicate as UpdateIssue — batch must not grow its own - // copy of the enqueue rule (the historical source of four-entry-point - // drift, MUL-3375). suppress_run applies batch-wide. if trigger, ok := h.IssueService.WillEnqueueRun(r.Context(), service.IssueTriggerInput{ Issue: issue, @@ -3449,23 +3587,10 @@ func (h *Handler) BatchUpdateIssues(w http.ResponseWriter, r *http.Request) { h.dispatchIssueRun(r.Context(), issue, trigger, actorType, actorID, req.Updates.HandoffNote) } - // No status change — not even → cancelled — cancels active tasks here, - // mirroring UpdateIssue (MUL-4465). See that handler for the rationale. - - // Platform-driven parent notification, mirrored from UpdateIssue - // (MUL-2538) but DEFERRED to after the loop. Evaluating the stage - // barrier here, per-child, would read a mid-batch sibling snapshot and - // fire a stale "advance Stage N+1" wake when one batch closes several - // stages at once (MUL-4155). Collect the terminal transitions and let - // notifyParentsOfBatchChildDone below evaluate each parent once against - // the batch's final committed state. Same transition guard as - // notifyParentOfChildDone: a non-terminal -> terminal move on a child. if statusChanged && issue.ParentIssueID.Valid && !isTerminalChildStatus(prevIssue.Status) && isTerminalChildStatus(issue.Status) { childDoneCompleted = append(childDoneCompleted, issue) } - - updated++ } // Aggregate parent/stage notification over the whole batch's final state so diff --git a/server/internal/handler/issue_child_done_test.go b/server/internal/handler/issue_child_done_test.go index d3949964a92..7bee91a9062 100644 --- a/server/internal/handler/issue_child_done_test.go +++ b/server/internal/handler/issue_child_done_test.go @@ -188,26 +188,45 @@ func TestChildReopenAndDoneFiresAgain(t *testing.T) { } } -// TestChildDoneSkippedWhenParentDone — when the parent is already at a -// terminal status, there is nothing for the parent assignee to advance to, -// so the notification must NOT fire. +// TestChildDoneSkippedWhenParentDone preserves the notification guard for a +// historic or non-HTTP inconsistent tree. The parent-state constraint now +// rejects this relationship before a user can create it, so exercise the +// notification helper directly instead of constructing an impossible API +// fixture. func TestChildDoneSkippedWhenParentDone(t *testing.T) { - fx := newChildDoneFixture(t, "done") - - updateChildStatus(t, fx.child.ID, "done") - - if got := countSystemCommentsOn(t, fx.parent.ID); got != 0 { + parent := createParentStateIssue(t, "child-done completed parent", "done", "") + parentRow, err := testHandler.Queries.GetIssue(context.Background(), parseUUID(parent.ID)) + if err != nil { + t.Fatalf("load completed parent: %v", err) + } + prev := parentRow + prev.Status = "in_progress" + child := parentRow + child.Status = "done" + child.ParentIssueID = parentRow.ID + testHandler.notifyParentOfChildDone(context.Background(), prev, child) + + if got := countSystemCommentsOn(t, parent.ID); got != 0 { t.Errorf("parent at 'done' should not receive notification, got %d comments", got) } } -// TestChildDoneSkippedWhenParentCancelled — same as above for cancelled. +// TestChildDoneSkippedWhenParentCancelled is the cancelled-parent counterpart +// of the historic-data guard above. func TestChildDoneSkippedWhenParentCancelled(t *testing.T) { - fx := newChildDoneFixture(t, "cancelled") - - updateChildStatus(t, fx.child.ID, "done") - - if got := countSystemCommentsOn(t, fx.parent.ID); got != 0 { + parent := createParentStateIssue(t, "child-done cancelled parent", "cancelled", "") + parentRow, err := testHandler.Queries.GetIssue(context.Background(), parseUUID(parent.ID)) + if err != nil { + t.Fatalf("load cancelled parent: %v", err) + } + prev := parentRow + prev.Status = "in_progress" + child := parentRow + child.Status = "done" + child.ParentIssueID = parentRow.ID + testHandler.notifyParentOfChildDone(context.Background(), prev, child) + + if got := countSystemCommentsOn(t, parent.ID); got != 0 { t.Errorf("parent at 'cancelled' should not receive notification, got %d comments", got) } } diff --git a/server/internal/handler/issue_parent_state_constraint_test.go b/server/internal/handler/issue_parent_state_constraint_test.go new file mode 100644 index 00000000000..9717dbc5e3e --- /dev/null +++ b/server/internal/handler/issue_parent_state_constraint_test.go @@ -0,0 +1,474 @@ +package handler + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/multica-ai/multica/server/internal/events" + "github.com/multica-ai/multica/server/pkg/protocol" +) + +const ( + parentHasIncompleteDescendantsCode = "parent_has_incomplete_descendants" + parentMustBeReopenedCode = "parent_must_be_reopened" +) + +type parentStateConflictResponse struct { + Code string `json:"code"` + Error string `json:"error"` + ParentIssueID string `json:"parent_issue_id"` + IncompleteDescendantCount *int `json:"incomplete_descendant_count,omitempty"` +} + +func createParentStateIssue(t *testing.T, title, status, parentID string) IssueResponse { + t.Helper() + w := httptest.NewRecorder() + body := map[string]any{ + "title": title + " " + time.Now().Format(time.RFC3339Nano), + "status": status, + } + if parentID != "" { + body["parent_issue_id"] = parentID + } + testHandler.CreateIssue(w, newRequest(http.MethodPost, "/api/issues?workspace_id="+testWorkspaceID, body)) + if w.Code != http.StatusCreated { + t.Fatalf("create %q: expected 201, got %d: %s", title, w.Code, w.Body.String()) + } + var issue IssueResponse + if err := json.NewDecoder(w.Body).Decode(&issue); err != nil { + t.Fatalf("decode created %q: %v", title, err) + } + t.Cleanup(func() { + testPool.Exec(context.Background(), `DELETE FROM issue WHERE id = $1`, issue.ID) + }) + return issue +} + +func updateParentStateIssue(t *testing.T, issueID string, body map[string]any) *httptest.ResponseRecorder { + t.Helper() + w := httptest.NewRecorder() + req := withURLParam(newRequest(http.MethodPut, "/api/issues/"+issueID, body), "id", issueID) + testHandler.UpdateIssue(w, req) + return w +} + +func decodeParentStateConflict(t *testing.T, w *httptest.ResponseRecorder, wantCode, wantParentID string) parentStateConflictResponse { + t.Helper() + if w.Code != http.StatusConflict { + t.Fatalf("expected 409 %s, got %d: %s", wantCode, w.Code, w.Body.String()) + } + var raw map[string]json.RawMessage + if err := json.Unmarshal(w.Body.Bytes(), &raw); err != nil { + t.Fatalf("decode raw conflict: %v", err) + } + for key := range raw { + switch key { + case "code", "error", "parent_issue_id", "incomplete_descendant_count": + default: + t.Fatalf("conflict response exposed unexpected field %q: %s", key, w.Body.String()) + } + } + var body parentStateConflictResponse + if err := json.NewDecoder(w.Body).Decode(&body); err != nil { + t.Fatalf("decode conflict: %v", err) + } + if body.Code != wantCode { + t.Fatalf("conflict code = %q, want %q: %#v", body.Code, wantCode, body) + } + if body.ParentIssueID != wantParentID { + t.Fatalf("conflict parent_issue_id = %q, want %q: %#v", body.ParentIssueID, wantParentID, body) + } + return body +} + +// TestParentStateConstraintBatchIsAtomicAndOrdersTreeTransitions proves that +// an invalid selected item rolls back earlier selected writes, while a valid +// parent-and-child completion succeeds regardless of request order. +func TestParentStateConstraintBatchIsAtomicAndOrdersTreeTransitions(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + parent := createParentStateIssue(t, "parent-state batch order parent", "in_review", "") + child := createParentStateIssue(t, "parent-state batch order child", "in_review", parent.ID) + w := httptest.NewRecorder() + testHandler.BatchUpdateIssues(w, newRequest(http.MethodPost, "/api/issues/batch-update", map[string]any{ + // The parent deliberately appears first. The handler must perform the + // terminal transition leaf-first inside its one transaction. + "issue_ids": []string{parent.ID, child.ID}, + "updates": map[string]any{"status": "done"}, + })) + if w.Code != http.StatusOK { + t.Fatalf("ordered batch completion: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := parentStateStatus(t, parent.ID); got != "done" { + t.Fatalf("ordered batch parent status = %q, want done", got) + } + if got := parentStateStatus(t, child.ID); got != "done" { + t.Fatalf("ordered batch child status = %q, want done", got) + } + + parentWithStaleSelection := createParentStateIssue(t, "parent-state batch stale parent", "in_review", "") + childWithStaleSelection := createParentStateIssue(t, "parent-state batch stale child", "in_review", parentWithStaleSelection.ID) + w = httptest.NewRecorder() + testHandler.BatchUpdateIssues(w, newRequest(http.MethodPost, "/api/issues/batch-update", map[string]any{ + // A stale selection in the middle must not stop the known child from + // being ordered before its parent. The stale ID remains skipped under + // the endpoint's existing batch contract. + "issue_ids": []string{parentWithStaleSelection.ID, "00000000-0000-0000-0000-000000000001", childWithStaleSelection.ID}, + "updates": map[string]any{"status": "done"}, + })) + if w.Code != http.StatusOK { + t.Fatalf("ordered batch with stale selection: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := parentStateStatus(t, parentWithStaleSelection.ID); got != "done" { + t.Fatalf("ordered stale batch parent status = %q, want done", got) + } + if got := parentStateStatus(t, childWithStaleSelection.ID); got != "done" { + t.Fatalf("ordered stale batch child status = %q, want done", got) + } + + completedParent := createParentStateIssue(t, "parent-state batch rollback parent", "in_review", "") + terminalChild := createParentStateIssue(t, "parent-state batch rollback child", "done", completedParent.ID) + if w := updateParentStateIssue(t, completedParent.ID, map[string]any{"status": "done"}); w.Code != http.StatusOK { + t.Fatalf("finish rollback parent: expected 200, got %d: %s", w.Code, w.Body.String()) + } + unrelated := createParentStateIssue(t, "parent-state batch rollback unrelated", "done", "") + w = httptest.NewRecorder() + testHandler.BatchUpdateIssues(w, newRequest(http.MethodPost, "/api/issues/batch-update", map[string]any{ + "issue_ids": []string{unrelated.ID, terminalChild.ID}, + "updates": map[string]any{"status": "todo"}, + })) + decodeParentStateConflict(t, w, parentMustBeReopenedCode, completedParent.ID) + if got := parentStateStatus(t, unrelated.ID); got != "done" { + t.Fatalf("conflicted batch partially changed unrelated issue to %q, want done", got) + } + if got := parentStateStatus(t, terminalChild.ID); got != "done" { + t.Fatalf("conflicted batch changed terminal child to %q, want done", got) + } +} + +func parentStateStatus(t *testing.T, issueID string) string { + t.Helper() + var status string + if err := testPool.QueryRow(context.Background(), `SELECT status FROM issue WHERE id = $1`, issueID).Scan(&status); err != nil { + t.Fatalf("read issue %s status: %v", issueID, err) + } + return status +} + +// TestParentStateConstraintRejectsTerminalParentWithIncompleteDescendant +// proves the user-visible core rule and that the rejected request does not +// partially change the parent row. The grandchild makes this a recursive, +// rather than direct-child-only, regression test. +func TestParentStateConstraintRejectsTerminalParentWithIncompleteDescendant(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + parent := createParentStateIssue(t, "parent-state recursive parent", "in_review", "") + child := createParentStateIssue(t, "parent-state recursive child", "in_review", parent.ID) + grandchild := createParentStateIssue(t, "parent-state recursive grandchild", "todo", child.ID) + + var commentCountBefore int + if err := testPool.QueryRow(context.Background(), `SELECT COUNT(*) FROM comment WHERE issue_id = $1`, parent.ID).Scan(&commentCountBefore); err != nil { + t.Fatalf("count parent comments before rejected completion: %v", err) + } + var eventMu sync.Mutex + parentUpdateEvents := 0 + testHandler.Bus.Subscribe(protocol.EventIssueUpdated, func(event events.Event) { + payload, ok := event.Payload.(map[string]any) + if !ok { + return + } + issue, ok := payload["issue"].(IssueResponse) + if ok && issue.ID == parent.ID { + eventMu.Lock() + parentUpdateEvents++ + eventMu.Unlock() + } + }) + + conflict := decodeParentStateConflict(t, + updateParentStateIssue(t, parent.ID, map[string]any{"status": "done"}), + parentHasIncompleteDescendantsCode, + parent.ID, + ) + if conflict.IncompleteDescendantCount == nil || *conflict.IncompleteDescendantCount != 2 { + t.Fatalf("incomplete descendant count = %#v, want 2", conflict.IncompleteDescendantCount) + } + if got := parentStateStatus(t, parent.ID); got != "in_review" { + t.Fatalf("rejected parent update changed status to %q, want in_review", got) + } + var commentCountAfter int + if err := testPool.QueryRow(context.Background(), `SELECT COUNT(*) FROM comment WHERE issue_id = $1`, parent.ID).Scan(&commentCountAfter); err != nil { + t.Fatalf("count parent comments after rejected completion: %v", err) + } + if commentCountAfter != commentCountBefore { + t.Fatalf("rejected parent completion changed comment count from %d to %d", commentCountBefore, commentCountAfter) + } + eventMu.Lock() + rejectedParentUpdateEvents := parentUpdateEvents + eventMu.Unlock() + if rejectedParentUpdateEvents != 0 { + t.Fatalf("rejected parent completion emitted %d issue update event(s)", rejectedParentUpdateEvents) + } + + if w := updateParentStateIssue(t, grandchild.ID, map[string]any{"status": "done"}); w.Code != http.StatusOK { + t.Fatalf("finish grandchild: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if w := updateParentStateIssue(t, child.ID, map[string]any{"status": "done"}); w.Code != http.StatusOK { + t.Fatalf("finish child: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if w := updateParentStateIssue(t, parent.ID, map[string]any{"status": "done"}); w.Code != http.StatusOK { + t.Fatalf("finish parent after terminal descendants: expected 200, got %d: %s", w.Code, w.Body.String()) + } +} + +// TestParentStateConstraintRejectsActiveChildMutationUnderDoneParent covers +// create, reparent, and reopen. An explicit parent move to Review +// (in_review) permits work again; the server never silently reopens it. +func TestParentStateConstraintRejectsActiveChildMutationUnderDoneParent(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + parent := createParentStateIssue(t, "parent-state completed parent", "in_review", "") + if w := updateParentStateIssue(t, parent.ID, map[string]any{"status": "done"}); w.Code != http.StatusOK { + t.Fatalf("finish parent: expected 200, got %d: %s", w.Code, w.Body.String()) + } + + w := httptest.NewRecorder() + testHandler.CreateIssue(w, newRequest(http.MethodPost, "/api/issues?workspace_id="+testWorkspaceID, map[string]any{ + "title": "parent-state blocked create " + time.Now().Format(time.RFC3339Nano), + "status": "todo", + "parent_issue_id": parent.ID, + })) + decodeParentStateConflict(t, w, parentMustBeReopenedCode, parent.ID) + + terminalChild := createParentStateIssue(t, "parent-state terminal child", "done", parent.ID) + root := createParentStateIssue(t, "parent-state reparent source", "todo", "") + decodeParentStateConflict(t, + updateParentStateIssue(t, root.ID, map[string]any{"parent_issue_id": parent.ID}), + parentMustBeReopenedCode, + parent.ID, + ) + if got := parentStateStatus(t, root.ID); got != "todo" { + t.Fatalf("rejected reparent changed source status to %q, want todo", got) + } + decodeParentStateConflict(t, + updateParentStateIssue(t, terminalChild.ID, map[string]any{"status": "todo"}), + parentMustBeReopenedCode, + parent.ID, + ) + if got := parentStateStatus(t, terminalChild.ID); got != "done" { + t.Fatalf("rejected reopen changed child status to %q, want done", got) + } + + if w := updateParentStateIssue(t, parent.ID, map[string]any{"status": "in_review"}); w.Code != http.StatusOK { + t.Fatalf("explicit parent reopen: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if w := updateParentStateIssue(t, root.ID, map[string]any{"parent_issue_id": parent.ID}); w.Code != http.StatusOK { + t.Fatalf("reparent after explicit reopen: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if w := updateParentStateIssue(t, terminalChild.ID, map[string]any{"status": "todo"}); w.Code != http.StatusOK { + t.Fatalf("reopen child after explicit parent reopen: expected 200, got %d: %s", w.Code, w.Body.String()) + } +} + +// TestParentStateConstraintRejectsBatchAndDirectWriters ensures neither the +// public batch endpoint nor a non-HTTP writer can bypass the same invariant. +func TestParentStateConstraintRejectsBatchAndDirectWriters(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + parent := createParentStateIssue(t, "parent-state batch parent", "in_review", "") + child := createParentStateIssue(t, "parent-state batch child", "done", parent.ID) + if w := updateParentStateIssue(t, parent.ID, map[string]any{"status": "done"}); w.Code != http.StatusOK { + t.Fatalf("finish parent: expected 200, got %d: %s", w.Code, w.Body.String()) + } + + w := httptest.NewRecorder() + testHandler.BatchUpdateIssues(w, newRequest(http.MethodPost, "/api/issues/batch-update", map[string]any{ + "issue_ids": []string{child.ID}, + "updates": map[string]any{"status": "todo"}, + })) + decodeParentStateConflict(t, w, parentMustBeReopenedCode, parent.ID) + if got := parentStateStatus(t, child.ID); got != "done" { + t.Fatalf("rejected batch reopen changed child status to %q, want done", got) + } + + _, err := testPool.Exec(context.Background(), `UPDATE issue SET status = 'todo' WHERE id = $1`, child.ID) + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Message != parentMustBeReopenedCode { + t.Fatalf("direct writer error = %v, want PostgreSQL %q", err, parentMustBeReopenedCode) + } + if got := parentStateStatus(t, child.ID); got != "done" { + t.Fatalf("rejected direct reopen changed child status to %q, want done", got) + } + + _, err = testPool.Exec(context.Background(), ` + INSERT INTO issue ( + workspace_id, title, status, priority, creator_type, creator_id, + parent_issue_id, position, number + ) + SELECT $1, $2, 'todo', 'none', 'member', $3, $4, 0, + COALESCE(MAX(number), 0) + 1 + FROM issue + WHERE workspace_id = $1 + GROUP BY $1 + `, testWorkspaceID, "parent-state direct insert "+time.Now().Format(time.RFC3339Nano), testUserID, parent.ID) + if !errors.As(err, &pgErr) || pgErr.Message != parentMustBeReopenedCode { + t.Fatalf("direct insert error = %v, want PostgreSQL %q", err, parentMustBeReopenedCode) + } +} + +// TestParentStateConstraintSerializesConcurrentParentDoneAndChildReopen is a +// two-connection regression test. One request can win, but the final tree +// cannot contain a done parent and an active child. +func TestParentStateConstraintSerializesConcurrentParentDoneAndChildReopen(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + parent := createParentStateIssue(t, "parent-state concurrent parent", "in_review", "") + child := createParentStateIssue(t, "parent-state concurrent child", "done", parent.ID) + + ctx := context.Background() + parentConn, err := testPool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire parent connection: %v", err) + } + defer parentConn.Release() + childConn, err := testPool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire child connection: %v", err) + } + defer childConn.Release() + + parentTx, err := parentConn.Begin(ctx) + if err != nil { + t.Fatalf("begin parent transaction: %v", err) + } + defer parentTx.Rollback(ctx) + if _, err := parentTx.Exec(ctx, `UPDATE issue SET status = 'done' WHERE id = $1`, parent.ID); err != nil { + t.Fatalf("parent completion setup: %v", err) + } + + childResult := make(chan error, 1) + go func() { + tx, err := childConn.Begin(ctx) + if err == nil { + _, err = tx.Exec(ctx, `UPDATE issue SET status = 'todo' WHERE id = $1`, child.ID) + if err == nil { + err = tx.Commit(ctx) + } else { + _ = tx.Rollback(ctx) + } + } + childResult <- err + }() + + select { + case err := <-childResult: + t.Fatalf("child reopen completed before parent transaction committed: %v", err) + case <-time.After(250 * time.Millisecond): + } + if err := parentTx.Commit(ctx); err != nil { + t.Fatalf("commit parent completion: %v", err) + } + if err := <-childResult; err == nil { + t.Fatal("child reopen succeeded after concurrent parent completion") + } + if got := parentStateStatus(t, parent.ID); got != "done" { + t.Fatalf("parent status = %q, want done", got) + } + if got := parentStateStatus(t, child.ID); got != "done" { + t.Fatalf("child status = %q, want done", got) + } +} + +// TestParentStateConstraintRechecksAncestorLocksAfterConcurrentReparent +// exercises the stale-lock-set race: while a descendant activation waits on a +// reparent, the trigger must discover and lock the newly committed ancestor +// before it validates. A third transaction must therefore be unable to take +// that new ancestor's advisory lock while the first update is still open. +func TestParentStateConstraintRechecksAncestorLocksAfterConcurrentReparent(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + + parent := createParentStateIssue(t, "parent-state recheck old parent", "in_review", "") + child := createParentStateIssue(t, "parent-state recheck child", "in_review", parent.ID) + grandchild := createParentStateIssue(t, "parent-state recheck grandchild", "done", child.ID) + newParent := createParentStateIssue(t, "parent-state recheck new parent", "in_review", "") + ctx := context.Background() + + reparentConn, err := testPool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire reparent connection: %v", err) + } + defer reparentConn.Release() + activateConn, err := testPool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire activation connection: %v", err) + } + defer activateConn.Release() + probeConn, err := testPool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire probe connection: %v", err) + } + defer probeConn.Release() + + reparentTx, err := reparentConn.Begin(ctx) + if err != nil { + t.Fatalf("begin reparent transaction: %v", err) + } + defer reparentTx.Rollback(ctx) + if _, err := reparentTx.Exec(ctx, `UPDATE issue SET parent_issue_id = $1 WHERE id = $2`, newParent.ID, child.ID); err != nil { + t.Fatalf("reparent child setup: %v", err) + } + + activateTx, err := activateConn.Begin(ctx) + if err != nil { + t.Fatalf("begin activation transaction: %v", err) + } + defer activateTx.Rollback(ctx) + activateResult := make(chan error, 1) + go func() { + _, err := activateTx.Exec(ctx, `UPDATE issue SET status = 'todo' WHERE id = $1`, grandchild.ID) + activateResult <- err + }() + select { + case err := <-activateResult: + t.Fatalf("activation completed before reparent committed: %v", err) + case <-time.After(250 * time.Millisecond): + } + if err := reparentTx.Commit(ctx); err != nil { + t.Fatalf("commit reparent: %v", err) + } + if err := <-activateResult; err != nil { + t.Fatalf("activation after legal reparent: %v", err) + } + + probeTx, err := probeConn.Begin(ctx) + if err != nil { + t.Fatalf("begin advisory-lock probe: %v", err) + } + defer probeTx.Rollback(ctx) + var acquired bool + if err := probeTx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock(hashtextextended($1::text, 0))`, newParent.ID).Scan(&acquired); err != nil { + t.Fatalf("probe new ancestor lock: %v", err) + } + if acquired { + t.Fatal("activation did not retain the newly reparented ancestor lock") + } +} diff --git a/server/internal/issueguard/parent_state.go b/server/internal/issueguard/parent_state.go new file mode 100644 index 00000000000..690b244da5f --- /dev/null +++ b/server/internal/issueguard/parent_state.go @@ -0,0 +1,62 @@ +package issueguard + +import ( + "errors" + "strconv" + "strings" + + "github.com/jackc/pgx/v5/pgconn" +) + +const ( + ParentHasIncompleteDescendantsCode = "parent_has_incomplete_descendants" + ParentMustBeReopenedCode = "parent_must_be_reopened" +) + +// ParentStateConflict is the transport-safe view of the database constraint. +// It carries only the affected parent ID and, when a parent completion was +// denied, the aggregate incomplete count; no descendant title or detail can +// leak through this error path. +type ParentStateConflict struct { + Code string + ParentIssueID string + IncompleteDescendantCount *int +} + +func ParentStateConflictFrom(err error) (ParentStateConflict, bool) { + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "P0001" { + return ParentStateConflict{}, false + } + if pgErr.Message != ParentHasIncompleteDescendantsCode && pgErr.Message != ParentMustBeReopenedCode { + return ParentStateConflict{}, false + } + + conflict := ParentStateConflict{Code: pgErr.Message} + for _, part := range strings.Split(pgErr.Detail, ";") { + key, value, ok := strings.Cut(part, "=") + if !ok { + continue + } + switch key { + case "parent_issue_id": + conflict.ParentIssueID = value + case "incomplete_descendant_count": + if count, err := strconv.Atoi(value); err == nil { + conflict.IncompleteDescendantCount = &count + } + } + } + return conflict, conflict.ParentIssueID != "" +} + +func ParentStateConflictMessage(code string) string { + switch code { + case ParentHasIncompleteDescendantsCode: + return "parent issue has incomplete descendants" + case ParentMustBeReopenedCode: + return "parent issue must be reopened before activating a child issue" + default: + return "parent issue state conflict" + } +} diff --git a/server/internal/service/task.go b/server/internal/service/task.go index 2900d42085c..65a40646711 100644 --- a/server/internal/service/task.go +++ b/server/internal/service/task.go @@ -18,6 +18,7 @@ import ( "github.com/multica-ai/multica/server/internal/attribution" "github.com/multica-ai/multica/server/internal/events" "github.com/multica-ai/multica/server/internal/featureflags" + "github.com/multica-ai/multica/server/internal/issueguard" obsmetrics "github.com/multica-ai/multica/server/internal/metrics" "github.com/multica-ai/multica/server/internal/realtime" "github.com/multica-ai/multica/server/internal/runtimeapps" @@ -3613,10 +3614,19 @@ func (s *TaskService) HandleFailedTasks(ctx context.Context, tasks []db.AgentTas WorkspaceID: issue.WorkspaceID, }) if updateErr != nil { - slog.Warn("handle failed tasks: reset stuck issue failed", - "issue_id", issueKey, - "error", updateErr, - ) + if conflict, ok := issueguard.ParentStateConflictFrom(updateErr); ok { + slog.Warn("handle failed tasks: reset rejected by parent-state constraint", + "issue_id", issueKey, + "workspace_id", workspaceID, + "parent_issue_id", conflict.ParentIssueID, + "conflict_code", conflict.Code, + ) + } else { + slog.Warn("handle failed tasks: reset stuck issue failed", + "issue_id", issueKey, + "error", updateErr, + ) + } } else { // This direct reset bypasses the HTTP UpdateIssue // handler that normally emits issue:updated, so emit diff --git a/server/migrations/202_issue_parent_state_constraint.down.sql b/server/migrations/202_issue_parent_state_constraint.down.sql new file mode 100644 index 00000000000..96c66fb753d --- /dev/null +++ b/server/migrations/202_issue_parent_state_constraint.down.sql @@ -0,0 +1,2 @@ +DROP TRIGGER IF EXISTS trg_issue_parent_state_constraint ON issue; +DROP FUNCTION IF EXISTS enforce_issue_parent_state_constraint(); diff --git a/server/migrations/202_issue_parent_state_constraint.up.sql b/server/migrations/202_issue_parent_state_constraint.up.sql new file mode 100644 index 00000000000..bab25b88d34 --- /dev/null +++ b/server/migrations/202_issue_parent_state_constraint.up.sql @@ -0,0 +1,143 @@ +-- Parent/child completion is a database invariant because issue rows are +-- written by HTTP handlers, background workers, integrations, and imports. +-- Keeping the verdict at the row-write boundary means no caller can observe a +-- completed parent with an active descendant by skipping an application-side +-- preflight check. +CREATE OR REPLACE FUNCTION enforce_issue_parent_state_constraint() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +DECLARE + old_parent_issue_id UUID; + lock_id UUID; + newly_locked_count INTEGER; + locked_ids UUID[] := ARRAY[]::UUID[]; + incomplete_descendant_count BIGINT; + terminal_parent_id UUID; +BEGIN + -- UpdateIssue always names status and parent_issue_id in its SQL, even for + -- title-only edits. Avoid taking hierarchy locks unless either value + -- actually changes. + IF TG_OP = 'UPDATE' + AND NEW.status IS NOT DISTINCT FROM OLD.status + AND NEW.parent_issue_id IS NOT DISTINCT FROM OLD.parent_issue_id THEN + RETURN NEW; + END IF; + + IF TG_OP = 'UPDATE' THEN + old_parent_issue_id := OLD.parent_issue_id; + END IF; + + -- Every competing operation locks the changed row plus both old and new + -- ancestor chains. A concurrent reparent can commit while this statement + -- waits on one of those locks, so recompute after each acquisition round + -- until no newly visible ancestor remains. Without this fixed-point loop, + -- a later terminal-parent update could miss the reparented branch. + -- + -- The first round is UUID-sorted. A newly discovered lower UUID can make + -- a later round contend out of order; PostgreSQL then aborts one contender + -- rather than permitting an inconsistent tree, and callers can retry it. + LOOP + newly_locked_count := 0; + FOR lock_id IN + WITH RECURSIVE related(id) AS ( + SELECT NEW.id + UNION + SELECT NEW.parent_issue_id + WHERE NEW.parent_issue_id IS NOT NULL + UNION + SELECT old_parent_issue_id + WHERE old_parent_issue_id IS NOT NULL + UNION + SELECT i.parent_issue_id + FROM issue AS i + JOIN related AS r ON r.id = i.id + WHERE i.parent_issue_id IS NOT NULL + ) + SELECT id + FROM related + WHERE NOT (id = ANY(locked_ids)) + ORDER BY id + LOOP + PERFORM pg_advisory_xact_lock(hashtextextended(lock_id::text, 0)); + locked_ids := array_append(locked_ids, lock_id); + newly_locked_count := newly_locked_count + 1; + END LOOP; + + EXIT WHEN newly_locked_count = 0; + END LOOP; + + -- A terminal parent is valid only when every direct or indirect child is + -- terminal. UNION makes a malformed historic cycle finite rather than + -- allowing the validation query itself to loop forever. + IF NEW.status IN ('done', 'cancelled') THEN + WITH RECURSIVE descendants(id) AS ( + SELECT id + FROM issue + WHERE parent_issue_id = NEW.id + UNION + SELECT child.id + FROM issue AS child + JOIN descendants AS descendant ON child.parent_issue_id = descendant.id + ) + SELECT COUNT(*) + INTO incomplete_descendant_count + FROM issue AS descendant_issue + JOIN descendants ON descendants.id = descendant_issue.id + WHERE descendant_issue.status NOT IN ('done', 'cancelled'); + + IF incomplete_descendant_count > 0 THEN + RAISE EXCEPTION USING + ERRCODE = 'P0001', + MESSAGE = 'parent_has_incomplete_descendants', + DETAIL = format( + 'parent_issue_id=%s;incomplete_descendant_count=%s', + NEW.id, + incomplete_descendant_count + ); + END IF; + END IF; + + -- Creating, reparenting, or reopening a non-terminal issue below any + -- terminal ancestor is forbidden. The caller must first make a separate, + -- explicit parent reopen (normally the Review/in_review state); this + -- trigger never mutates the parent on the caller's behalf. + IF NEW.status NOT IN ('done', 'cancelled') + AND NEW.parent_issue_id IS NOT NULL THEN + WITH RECURSIVE ancestors(id, parent_issue_id, status, depth, path) AS ( + SELECT id, parent_issue_id, status, 1, ARRAY[id] + FROM issue + WHERE id = NEW.parent_issue_id + UNION ALL + SELECT parent.id, + parent.parent_issue_id, + parent.status, + ancestor.depth + 1, + ancestor.path || parent.id + FROM issue AS parent + JOIN ancestors AS ancestor ON parent.id = ancestor.parent_issue_id + WHERE NOT parent.id = ANY(ancestor.path) + ) + SELECT id + INTO terminal_parent_id + FROM ancestors + WHERE status IN ('done', 'cancelled') + ORDER BY depth + LIMIT 1; + + IF terminal_parent_id IS NOT NULL THEN + RAISE EXCEPTION USING + ERRCODE = 'P0001', + MESSAGE = 'parent_must_be_reopened', + DETAIL = format('parent_issue_id=%s', terminal_parent_id); + END IF; + END IF; + + RETURN NEW; +END; +$$; + +CREATE TRIGGER trg_issue_parent_state_constraint +BEFORE INSERT OR UPDATE OF status, parent_issue_id ON issue +FOR EACH ROW +EXECUTE FUNCTION enforce_issue_parent_state_constraint();