Integrate the shared document renderer and two-trees project work #209
@@ -430,6 +430,21 @@ func refuseCandidateRawContextScope(db *sql.DB) error {
|
||||
if columns > 0 {
|
||||
return fmt.Errorf("candidate context scope text columns remain; isolate and scrub source before backup or migration")
|
||||
}
|
||||
var version int
|
||||
if sqliteHasTable(db, "schema_migrations") {
|
||||
if err := db.QueryRow(`SELECT COALESCE(MAX(version),0) FROM schema_migrations`).Scan(&version); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if version >= 34 {
|
||||
if !sqliteHasTable(db, "context_scope_scrub_state") {
|
||||
return fmt.Errorf("candidate context physical scrub marker missing; isolate and scrub source before backup or migration")
|
||||
}
|
||||
var pending int
|
||||
if err := db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 0 {
|
||||
return fmt.Errorf("candidate context physical scrub pending or unverified; isolate and scrub source before backup or migration")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -173,7 +173,8 @@ func TestCopyOrderCoversAllFixtureTables(t *testing.T) {
|
||||
for _, t := range copyOrder {
|
||||
known[t] = true
|
||||
}
|
||||
known["schema_migrations"] = true // 迁移器管理,不随迁
|
||||
known["schema_migrations"] = true // 迁移器管理,不随迁
|
||||
known["context_scope_scrub_state"] = true // SQLite 候选原文物理清理状态,不随迁
|
||||
// 运行态表不随迁:执行许可是进程瞬态;执行租约与跨实例命令属当前
|
||||
// 实例的调度现场(接管语义在 U5 内由 lease/fencing 处理,不复制历史)。
|
||||
for _, rt := range []string{"admission_leases", "conversation_executions", "run_commands"} {
|
||||
@@ -248,3 +249,58 @@ func TestBackupAndMigrateRejectCandidateRawContextScope(t *testing.T) {
|
||||
t.Fatal("backup retained candidate raw scope body")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupAndMigrateRejectPendingOrMissingContextScrubMarker(t *testing.T) {
|
||||
path := buildFixture(t)
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, query := range []string{
|
||||
`CREATE TABLE IF NOT EXISTS context_scope_scrub_state(id INTEGER PRIMARY KEY,pending INTEGER NOT NULL)`,
|
||||
`INSERT OR REPLACE INTO context_scope_scrub_state(id,pending) VALUES(1,1)`,
|
||||
} {
|
||||
if _, err := db.Exec(query); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertRejected := func(name string) {
|
||||
t.Helper()
|
||||
out := filepath.Join(t.TempDir(), name+".db")
|
||||
if err := cmdBackup([]string{"--source", path, "--out", out}); err == nil {
|
||||
t.Fatalf("backup accepted %s context scrub state", name)
|
||||
}
|
||||
if _, err := os.Stat(out); !os.IsNotExist(err) {
|
||||
t.Fatalf("rejected %s backup created output: %v", name, err)
|
||||
}
|
||||
if err := cmdMigrate([]string{"--source", path, "--target", "postgres://invalid", "--migration-id", name}); err == nil {
|
||||
t.Fatalf("migration accepted %s context scrub state", name)
|
||||
}
|
||||
}
|
||||
assertRejected("pending")
|
||||
db, err = sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.Exec(`DROP TABLE context_scope_scrub_state`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertRejected("missing")
|
||||
db, err = sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.Exec(`DELETE FROM schema_migrations WHERE version=35`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertRejected("v34-missing")
|
||||
}
|
||||
|
||||
@@ -128,6 +128,95 @@ func TestCandidateV34RawScopeColumnsAreScrubbedOnUpgrade(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCandidateV34CommittedDropKeepsPendingScrubAcrossRestart(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "candidate-crash.db")
|
||||
st, err := OpenSQLite(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "crash-model", InputSHA256: "input", Delivery: "attempted"}
|
||||
if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Bound: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, query := range []string{
|
||||
`DELETE FROM schema_migrations WHERE version=35`,
|
||||
`ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`,
|
||||
`ALTER TABLE context_manifests ADD COLUMN resolve_target_id TEXT NOT NULL DEFAULT ''`,
|
||||
`UPDATE context_manifests SET resolve_goal='OLD_RESTRICTED_BODY',resolve_target_id='doc-private'`,
|
||||
} {
|
||||
if _, err := st.db.Exec(query); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// Apply only the transactional DDL and ledger. A process crash at this
|
||||
// boundary must not make the next open mistake the file for scrubbed.
|
||||
if err := st.migratePass(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var pending int
|
||||
if err := st.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 1 {
|
||||
t.Fatalf("committed v35 lost physical scrub obligation: pending=%d err=%v", pending, err)
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
restarted, err := OpenSQLite(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer restarted.Close()
|
||||
if err := restarted.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 0 {
|
||||
t.Fatalf("restart did not complete pending scrub: pending=%d err=%v", pending, err)
|
||||
}
|
||||
for _, suffix := range []string{"", "-wal"} {
|
||||
data, err := os.ReadFile(path + suffix)
|
||||
if err == nil && bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) {
|
||||
t.Fatalf("restart left candidate raw bytes in SQLite %s", suffix)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestEarlierV35WithoutMarkerIsConservativelyRescrubbed(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "old-v35.db")
|
||||
st, err := OpenSQLite(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "old-v35-model", InputSHA256: "input", Delivery: "attempted"}
|
||||
if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Bound: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, query := range []string{
|
||||
`ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`,
|
||||
`UPDATE context_manifests SET resolve_goal='OLD_RESTRICTED_BODY'`,
|
||||
`UPDATE context_manifests SET resolve_goal=''`,
|
||||
`ALTER TABLE context_manifests DROP COLUMN resolve_goal`,
|
||||
`DROP TABLE context_scope_scrub_state`,
|
||||
} {
|
||||
if _, err := st.db.Exec(query); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
restarted, err := OpenSQLite(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer restarted.Close()
|
||||
var pending int
|
||||
if err := restarted.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil || pending != 0 {
|
||||
t.Fatalf("earlier v35 without marker was not re-scrubbed: pending=%d err=%v", pending, err)
|
||||
}
|
||||
for _, suffix := range []string{"", "-wal"} {
|
||||
data, err := os.ReadFile(path + suffix)
|
||||
if err == nil && bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) {
|
||||
t.Fatalf("old v35 raw bytes survived re-scrub in SQLite %s", suffix)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCandidateV34PostgresRawScopeScrubAndFreshSchema(t *testing.T) {
|
||||
st := openTestStore(t)
|
||||
if st.Dialect() != "postgres" {
|
||||
|
||||
+57
-13
@@ -881,8 +881,11 @@ func init() {
|
||||
},
|
||||
},
|
||||
migration{version: 35, name: "scrub_candidate_context_scope_text",
|
||||
up: map[string][]string{"sqlite": {`SELECT 1`}, "postgres": {`SELECT 1`}},
|
||||
down: map[string][]string{"sqlite": {`SELECT 1`}, "postgres": {`SELECT 1`}},
|
||||
up: map[string][]string{
|
||||
"sqlite": {`CREATE TABLE IF NOT EXISTS context_scope_scrub_state(id INTEGER PRIMARY KEY CHECK(id=1),pending INTEGER NOT NULL CHECK(pending IN (0,1)))`},
|
||||
"postgres": {`SELECT 1`},
|
||||
},
|
||||
down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS context_scope_scrub_state`}, "postgres": {`SELECT 1`}},
|
||||
},
|
||||
)
|
||||
}
|
||||
@@ -976,17 +979,17 @@ func (s *Store) migrate() error {
|
||||
if s.dialect.name() == "postgres" {
|
||||
return s.migratePG()
|
||||
}
|
||||
var candidateRawColumns int
|
||||
if err := s.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&candidateRawColumns); err != nil {
|
||||
return err
|
||||
}
|
||||
// SQLite:_txlock=immediate(OpenSQLite DSN)保证迁移事务在 BEGIN 即取写锁,
|
||||
// 跨进程迁移者按 busy_timeout 排队;后续版本读取发生在同一事务内,
|
||||
// 前一个迁移者提交后本进程看到的已是完整账本(无部分版本可见)。
|
||||
if err := s.migratePass(); err != nil {
|
||||
return err
|
||||
}
|
||||
if candidateRawColumns > 0 {
|
||||
pending, err := s.ensureContextScrubState()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if pending {
|
||||
// The prior candidate schema may have stored full prompts. VACUUM is
|
||||
// outside the DDL transaction; it compacts the isolated SQLite file
|
||||
// after v35 drops those columns. Refuse to serve if compaction fails.
|
||||
@@ -1003,10 +1006,38 @@ func (s *Store) migrate() error {
|
||||
if busy != 0 {
|
||||
return errors.New("candidate context scrub checkpoint remains busy")
|
||||
}
|
||||
if _, err := s.db.Exec(`UPDATE context_scope_scrub_state SET pending=0 WHERE id=1 AND pending=1`); err != nil {
|
||||
return fmt.Errorf("candidate context scrub completion: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ensureContextScrubState handles databases that ran the earlier candidate
|
||||
// v35 before a durable marker existed. They are conservatively re-scrubbed.
|
||||
func (s *Store) ensureContextScrubState() (bool, error) {
|
||||
var rawColumns int
|
||||
if err := s.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&rawColumns); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if rawColumns != 0 {
|
||||
return false, errors.New("candidate context raw scope columns remain after migration")
|
||||
}
|
||||
if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS context_scope_scrub_state(id INTEGER PRIMARY KEY CHECK(id=1),pending INTEGER NOT NULL CHECK(pending IN (0,1)))`); err != nil {
|
||||
return false, err
|
||||
}
|
||||
// An absent row may be a partially initialized earlier candidate; mark it
|
||||
// pending before any physical cleanup. A fresh v35 inserted pending=0.
|
||||
if _, err := s.db.Exec(`INSERT OR IGNORE INTO context_scope_scrub_state(id,pending) VALUES(1,1)`); err != nil {
|
||||
return false, err
|
||||
}
|
||||
var pending int
|
||||
if err := s.db.QueryRow(`SELECT pending FROM context_scope_scrub_state WHERE id=1`).Scan(&pending); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return pending != 0, nil
|
||||
}
|
||||
|
||||
// migratePG 以会话级咨询锁串行化并发迁移者(KTD8):锁在专用连接上获取,
|
||||
// 等锁者在轮到自己后重新读取账本——前一个迁移者已完成时本轮为空操作。
|
||||
// 锁与迁移事务分离:迁移语句走连接池,锁只负责让 migrator 全局互斥。
|
||||
@@ -1082,8 +1113,10 @@ func (s *Store) migratePass() error {
|
||||
if applied[m.version] {
|
||||
continue
|
||||
}
|
||||
candidateRawScope := false
|
||||
if m.version == 35 {
|
||||
if err := scrubCandidateContextScope(tx, s.dialect.name()); err != nil {
|
||||
candidateRawScope, err = scrubCandidateContextScope(tx, s.dialect.name())
|
||||
if err != nil {
|
||||
return fmt.Errorf("migrate v35: %w", err)
|
||||
}
|
||||
}
|
||||
@@ -1096,6 +1129,15 @@ func (s *Store) migratePass() error {
|
||||
return fmt.Errorf("migrate v%d (%s): %w", m.version, m.name, err)
|
||||
}
|
||||
}
|
||||
if m.version == 35 && s.dialect.name() == "sqlite" {
|
||||
pending := 0
|
||||
if candidateRawScope {
|
||||
pending = 1
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT OR REPLACE INTO context_scope_scrub_state(id,pending) VALUES(1,?)`, pending); err != nil {
|
||||
return fmt.Errorf("migrate v35 scrub marker: %w", err)
|
||||
}
|
||||
}
|
||||
// 故障注入点:DDL 已执行、版本未登记——历史实现在此处崩溃会留下
|
||||
// 「DDL 生效而账本缺失」的中间态;同事务设计下这里失败必须整体回滚。
|
||||
if fault := s.takeMigrateFault(); fault != nil {
|
||||
@@ -1111,7 +1153,8 @@ func (s *Store) migratePass() error {
|
||||
// Candidate v34 was never released, but isolated test databases may already
|
||||
// contain raw Resolve goals. Fresh v34 creates no such columns; v35 removes
|
||||
// them conditionally from the earlier candidate schema before serving.
|
||||
func scrubCandidateContextScope(tx *sql.Tx, dialect string) error {
|
||||
func scrubCandidateContextScope(tx *sql.Tx, dialect string) (bool, error) {
|
||||
found := false
|
||||
for _, column := range []string{"resolve_goal", "resolve_target_id"} {
|
||||
var exists int
|
||||
var err error
|
||||
@@ -1121,19 +1164,20 @@ func scrubCandidateContextScope(tx *sql.Tx, dialect string) error {
|
||||
err = tx.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name=$1`, column).Scan(&exists)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
return false, err
|
||||
}
|
||||
if exists == 0 {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if _, err := tx.Exec(`UPDATE context_manifests SET ` + column + `=''`); err != nil {
|
||||
return err
|
||||
return false, err
|
||||
}
|
||||
if _, err := tx.Exec(`ALTER TABLE context_manifests DROP COLUMN ` + column); err != nil {
|
||||
return err
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return found, nil
|
||||
}
|
||||
|
||||
func latestMigrationVersion() int {
|
||||
|
||||
Reference in New Issue
Block a user