diff --git a/.doc/ops/long-term/2026-09-19-audit-remainder.md b/.doc/ops/long-term/2026-09-19-audit-remainder.md index affdf742..e6332529 100644 --- a/.doc/ops/long-term/2026-09-19-audit-remainder.md +++ b/.doc/ops/long-term/2026-09-19-audit-remainder.md @@ -45,10 +45,11 @@ updated: 2026-09-19 | H3 负例 | done。`TestListChatSessions_CrossUserIs403` | | FE 缓存 | done(版本字段)。窗口缓存写入 `{schema,data}`,无 schema 的旧载荷丢弃。`window cache: 无 schema` | | FE lucide | 关闭理由:`package-lock.json` 锁定 `lucide-react@1.42.0`,来源 `registry.npmjs.org`,带 integrity。`^1.16.0` 是合法的 1.x 范围,不是来源不明的包 | +| M7 | done(出站意图 + 同键重放)。`document_content_projections` 与账本同一事务写入;投影成功才销账。`TestDocRevision_PatchOrder_RetryProjectsWithoutSecondLedgerCommit`。账本库与正文库仍不是一个 SQL 事务,进程退出后靠未投影行重放,不靠客户端记忆 | | FE App.tsx | 仍开放。4600 行拆 hook 是独立重构,本轮未拆,避免在没有行为变更需求时改写工作台 | | H1-del | 关闭理由:`allowLegacyTestPrincipal` 仍被 `AllowInMemory && ContextDbHost==""` 的 dev/test 内存测试使用。删除会把这批测试变成未登录 401。保留告警,删除另开、先改测试夹具 | -仍开放:M7 修订三阶段事务、FE `App.tsx` 拆分、rdev S2/S3 与 P2–P4、首次加载 U1/U3–U7、晋升前置(dev rebuild、治理 PG、`npm test`/`npm run build`)。跨用户会话列表 403 已有 `TestListChatSessions_CrossUserIs403`。`CURRENT.md` 文首 test schema 已改为 52。 +仍开放:FE `App.tsx` 拆分、rdev S2/S3 与 P2–P4、首次加载 U1/U3–U7、晋升前置(dev rebuild、治理 PG、`npm test`/`npm run build`)。M7 的投影意图已随账本提交(goose `00053`);容器 schema 仍是 52,要等占槽重建后才升到 53。 红线关闭(不是实现): @@ -66,7 +67,7 @@ updated: 2026-09-19 | M4 | `fmt.Sscanf` 解析 `u_sess_*` 得到 UserID=0 | 同上 | 统一数字 ID | | M5 | `Runtime.Start` 双调用且吞错 | `cmd/server/main.go` 与 `server.go` | `sync.Once` 或删 Server.Start 内调用 | | M6 | sessions 无硬上限 | `server.go` cleanStaleSessionsLocked | LRU + 服务端发 session id | -| M7 | 修订三阶段写无事务 | `document_revision_handlers.go` | outbox 或单事务投影 | +| M7 | 修订三阶段写无事务 | `document_revision_handlers.go` | 已落地:`00053` 投影意图与账本同事务;成功后销账。见进展表 | | M9 | `SEMANTIC_SUMMARY_REPLACE` 只在 dev compose 声明;dev/test 默认模型不一致 | compose | 删未接线 flag 并钉死测试;确认 M3 是否刻意 | | M10 | cutover 只允许 `dev` | `internal/events/cutover.go` | 写明 test/main 切换路径 | | M13 | `deploy/postgres/migrations/` 死目录 | 与 goose 重复 | 确认无外部依赖后归档 | diff --git a/deploy/postgres/goose/00053_2026-09-19-document-content-projection.sql b/deploy/postgres/goose/00053_2026-09-19-document-content-projection.sql new file mode 100644 index 00000000..e4f45679 --- /dev/null +++ b/deploy/postgres/goose/00053_2026-09-19-document-content-projection.sql @@ -0,0 +1,22 @@ +-- +goose Up +-- 人工修订账本与正文投影不是同一个存储。投影意图与账本同一事务写入, +-- 投影成功后才标 projected。进程在两次写之间退出时,未投影行仍可重放。 + +CREATE TABLE IF NOT EXISTS document_content_projections ( + doc_id TEXT NOT NULL, + snapshot_id BIGINT NOT NULL, + revision_no INT NOT NULL, + idempotency_key TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'pending', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + projected_at TIMESTAMPTZ, + PRIMARY KEY (doc_id, snapshot_id), + CONSTRAINT chk_doc_content_projection_status CHECK (status IN ('pending', 'projected')) +); + +CREATE INDEX IF NOT EXISTS idx_doc_content_projection_pending + ON document_content_projections (created_at) + WHERE status = 'pending'; + +-- +goose Down +DROP TABLE IF EXISTS document_content_projections; diff --git a/internal/docrevision/pg_mutate.go b/internal/docrevision/pg_mutate.go index 28d49f50..a15d63a3 100644 --- a/internal/docrevision/pg_mutate.go +++ b/internal/docrevision/pg_mutate.go @@ -68,6 +68,13 @@ func (s *Service) openTemplatePGLocked(ctx context.Context, docID, canon, sha st func (s *Service) persistHumanRevisionLocked(ctx context.Context, root *DocRootPointers, ver *Version, snap *Snapshot, ev *ChangeEvent, isBaseline bool) error { if !s.usingPG() { + if root != nil && snap != nil { + key := "" + if ev != nil { + key = ev.IdempotencyKey + } + s.noteContentProjectionLocked(root.DocID, snap.ID, root.HeadRevisionNo, key) + } return nil } blob := s.blobs[snap.ContentSHA256] @@ -113,10 +120,29 @@ func (s *Service) persistHumanRevisionLocked(ctx context.Context, root *DocRootP return err } s.adoptEventID(ev, eid) + key := "" + if ev != nil { + key = ev.IdempotencyKey + } + if err := s.insertContentProjectionTX(ctx, tx, root.DocID, snap.ID, root.HeadRevisionNo, key); err != nil { + return err + } + s.noteContentProjectionLocked(root.DocID, snap.ID, root.HeadRevisionNo, key) return s.projectRootPointersTX(ctx, tx, root) }) } +func (s *Service) insertContentProjectionTX(ctx context.Context, tx pgx.Tx, docID string, snapshotID int64, revisionNo int, idempotencyKey string) error { + if docID == "" || snapshotID == 0 { + return nil + } + _, err := tx.Exec(ctx, ` + INSERT INTO document_content_projections (doc_id, snapshot_id, revision_no, idempotency_key, status) + VALUES ($1, $2, $3, $4, 'pending') + ON CONFLICT (doc_id, snapshot_id) DO NOTHING`, docID, snapshotID, revisionNo, idempotencyKey) + return err +} + func (s *Service) persistAgentCandidateLocked(ctx context.Context, outSnap *Snapshot, rep *ChangeReport, c *AgentCandidate) error { if !s.usingPG() { return nil diff --git a/internal/docrevision/projection_outbox.go b/internal/docrevision/projection_outbox.go new file mode 100644 index 00000000..e4ed2c6d --- /dev/null +++ b/internal/docrevision/projection_outbox.go @@ -0,0 +1,61 @@ +package docrevision + +import "context" + +// ContentProjection is the fact-source write still owed after a human ledger commit. +// The row is inserted in the same transaction as the ledger. It is removed only after +// DocGov / workspace / freeform projection succeeds. +type ContentProjection struct { + DocID string + SnapshotID int64 + RevisionNo int + IdempotencyKey string +} + +func (s *Service) noteContentProjectionLocked(docID string, snapshotID int64, revisionNo int, idempotencyKey string) { + if s == nil || docID == "" || snapshotID == 0 { + return + } + if s.projections == nil { + s.projections = map[string]ContentProjection{} + } + s.projections[docID] = ContentProjection{ + DocID: docID, + SnapshotID: snapshotID, + RevisionNo: revisionNo, + IdempotencyKey: idempotencyKey, + } +} + +// ContentProjectionPending reports an unprojected ledger head for docID. +func (s *Service) ContentProjectionPending(docID string) (ContentProjection, bool) { + if s == nil { + return ContentProjection{}, false + } + s.mu.Lock() + defer s.mu.Unlock() + p, ok := s.projections[docID] + return p, ok +} + +// MarkContentProjected drops the in-memory intent and acks the durable row. +// A snapshot mismatch leaves a newer pending head alone. +func (s *Service) MarkContentProjected(ctx context.Context, docID string, snapshotID int64) error { + if s == nil || docID == "" || snapshotID == 0 { + return nil + } + s.mu.Lock() + if p, ok := s.projections[docID]; ok && p.SnapshotID == snapshotID { + delete(s.projections, docID) + } + usePG := s.usingPG() + s.mu.Unlock() + if !usePG { + return nil + } + _, err := s.pool.Exec(ctx, ` + UPDATE document_content_projections + SET status = 'projected', projected_at = now() + WHERE doc_id = $1 AND snapshot_id = $2 AND status = 'pending'`, docID, snapshotID) + return err +} diff --git a/internal/docrevision/service.go b/internal/docrevision/service.go index 987694df..5e0cdd22 100644 --- a/internal/docrevision/service.go +++ b/internal/docrevision/service.go @@ -19,14 +19,15 @@ type Service struct { pool *pgxpool.Pool mu sync.Mutex - blobs map[string]*ContentBlob - snapshots map[int64]*Snapshot - versions map[int64]*Version - events map[int64]*ChangeEvent - candidates map[int64]*AgentCandidate - reports map[int64]*ChangeReport - roots map[string]*DocRootPointers - idemp map[string]int64 // idempotency_key -> event id + blobs map[string]*ContentBlob + snapshots map[int64]*Snapshot + versions map[int64]*Version + events map[int64]*ChangeEvent + candidates map[int64]*AgentCandidate + reports map[int64]*ChangeReport + roots map[string]*DocRootPointers + idemp map[string]int64 // idempotency_key -> event id + projections map[string]ContentProjection nextSnapID int64 nextVerID int64 @@ -71,6 +72,7 @@ func NewService(pool *pgxpool.Pool) *Service { reports: map[int64]*ChangeReport{}, roots: map[string]*DocRootPointers{}, idemp: map[string]int64{}, + projections: map[string]ContentProjection{}, LedgerWrite: true, LedgerRead: true, AgentCommit: true, @@ -1079,6 +1081,7 @@ func (s *Service) forgetDocLocked(docID string) { delete(s.reports, id) } } + delete(s.projections, docID) } func (s *Service) manualOverlayLocked(docID string) DiffResult { diff --git a/internal/server/document_revision_handlers.go b/internal/server/document_revision_handlers.go index 489989c6..9001b19d 100644 --- a/internal/server/document_revision_handlers.go +++ b/internal/server/document_revision_handlers.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "net/http" "strconv" "strings" @@ -270,9 +271,10 @@ func (s *Server) handleDocRevisionPatchContent(w http.ResponseWriter, r *http.Re return } - // 3) Project fact-source AFTER ledger commit. Controlled docs use SaveDocumentContent so - // author_last_edited / audit trail stay consistent; failure → DOCUMENT_PROJECTION_FAILED - // (ledger remains authoritative; same Idempotency-Key retry re-projects). + // 3) Project fact-source AFTER ledger commit. The projection intent was inserted in the + // same transaction as the ledger (document_content_projections). Failure keeps that row + // pending and returns DOCUMENT_PROJECTION_FAILED. The same Idempotency-Key retries + // projection without a second revision. Only a successful projection acks the row. if kind == "controlled" && s.gov != nil { if _, err := s.gov.SaveDocumentContent(r.Context(), docgov.ContentSaveRequest{ DocID: docID, @@ -315,6 +317,11 @@ func (s *Server) handleDocRevisionPatchContent(w http.ResponseWriter, r *http.Re return } } + if res.State.HeadSnapshotID != 0 { + if markErr := rev.MarkContentProjected(r.Context(), docID, res.State.HeadSnapshotID); markErr != nil { + slog.Warn("content projection ack failed", "doc_id", docID, "snapshot_id", res.State.HeadSnapshotID, "err", markErr) + } + } writeJSONResponse(w, http.StatusOK, res) } diff --git a/internal/server/document_revision_patch_order_test.go b/internal/server/document_revision_patch_order_test.go index 8875e86b..583535c0 100644 --- a/internal/server/document_revision_patch_order_test.go +++ b/internal/server/document_revision_patch_order_test.go @@ -222,3 +222,76 @@ func assertHumanLeaseReleasedSquadCanAcquire(t *testing.T, srv *Server, docID st t.Fatal(err) } } + +func TestDocRevision_PatchOrder_RetryProjectsWithoutSecondLedgerCommit(t *testing.T) { + srv := newTestServer(t) + ctx := context.Background() + dataRoot := srv.dataRoot() + blocker := filepath.Join(dataRoot, "proj-retry-block") + if err := os.WriteFile(blocker, []byte("not-a-dir"), 0o644); err != nil { + t.Fatal(err) + } + workDir := filepath.Join(blocker, "ws-retry") + docID := "doc_proj1_patch_retry" + + if _, err := srv.gov.CreateOrGetDocument(ctx, docID, 1, "intake", "摸底", "V0.1", workDir, 1, "admin", 2, "r", 3, "a"); err != nil { + t.Fatal(err) + } + st, err := srv.rev.OpenTemplateVersion(ctx, docID, "# base\n\nbody\n", docrevision.ActorSnapshot{DisplayName: "t"}, 1, workDir) + if err != nil { + t.Fatal(err) + } + + token := loginTestAdmin(t, srv) + body := `{"markdown":"# base\n\nbody\n\nedited by human\n"}` + patch := func() *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPatch, "/api/v1/documents/"+docID+"/content", bytes.NewReader([]byte(body))) + withBearerAuth(req, token) + req.Header.Set("Idempotency-Key", "patch-retry-k1") + req.Header.Set("If-Match", `"`+strconv.FormatInt(st.HeadSnapshotID, 10)+":"+strconv.Itoa(st.HeadRevisionNo)+`"`) + w := httptest.NewRecorder() + srv.Routes().ServeHTTP(w, req) + return w + } + + first := patch() + if first.Code != http.StatusInternalServerError { + t.Fatalf("first projection want 500 got %d %s", first.Code, first.Body.String()) + } + pending, ok := srv.rev.ContentProjectionPending(docID) + if !ok { + t.Fatal("ledger commit must leave a content projection pending") + } + afterFail, err := srv.rev.GetRevisionState(ctx, docID) + if err != nil { + t.Fatal(err) + } + if afterFail.HeadSnapshotID != pending.SnapshotID { + t.Fatalf("pending snapshot %d != ledger head %d", pending.SnapshotID, afterFail.HeadSnapshotID) + } + + if err := os.Remove(blocker); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(workDir, 0o755); err != nil { + t.Fatal(err) + } + second := patch() + if second.Code != http.StatusOK { + t.Fatalf("retry want 200 got %d %s", second.Code, second.Body.String()) + } + afterRetry, err := srv.rev.GetRevisionState(ctx, docID) + if err != nil { + t.Fatal(err) + } + if afterRetry.HeadRevisionNo != afterFail.HeadRevisionNo || afterRetry.HeadSnapshotID != afterFail.HeadSnapshotID { + t.Fatalf("retry must not write a second revision: fail=%+v retry=%+v", afterFail, afterRetry) + } + if _, still := srv.rev.ContentProjectionPending(docID); still { + t.Fatal("successful projection must ack the pending row") + } + if _, err := os.Stat(filepath.Join(workDir, "output.md")); err != nil { + t.Fatalf("retry must write output.md: %v", err) + } +}