fix(docrevision): ack content projection only after the fact source is written
The human ledger and the document body are different stores. Record the projection intent in the same transaction as the ledger, and clear it only when the body write succeeds so a retry cannot append a second revision. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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 重复 | 确认无外部依赖后归档 |
|
||||
|
||||
@@ -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;
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user