agent-exec transport e22dcba544696b290bc056e1a51e4132/df1c90387b2b
This commit is contained in:
@@ -309,6 +309,14 @@ var copyOrder = []string{
|
||||
"instance_guard",
|
||||
}
|
||||
|
||||
// schemaSeededSingletons:schema 迁移在每个新库上预置的单行表。它们不是目标
|
||||
// 业务数据,复制时须由源库的行替换。
|
||||
var schemaSeededSingletons = map[string]bool{
|
||||
"account_directory_revision": true,
|
||||
"authority_state": true,
|
||||
"role_authority_import_state": true,
|
||||
}
|
||||
|
||||
const controlDDL = `
|
||||
CREATE TABLE IF NOT EXISTS migration_control (
|
||||
migration_id TEXT PRIMARY KEY,
|
||||
@@ -517,9 +525,9 @@ func guardIdentity(db *sql.DB, migrationID, snapID string, srcVersion int) error
|
||||
func businessRows(db *sql.DB) (int64, error) {
|
||||
var total int64
|
||||
for _, t := range copyOrder {
|
||||
// Schema setup seeds this singleton before a migration starts. It is
|
||||
// replaced from the source during copying and is not target business data.
|
||||
if t == "account_directory_revision" {
|
||||
// Schema setup seeds these singletons before a migration starts. They are
|
||||
// replaced from the source during copying and are not target business data.
|
||||
if schemaSeededSingletons[t] {
|
||||
continue
|
||||
}
|
||||
var n int64
|
||||
@@ -799,11 +807,12 @@ func copyRowsInTx(tx *sql.Tx, src *sql.DB, table string, cols []string, bools ma
|
||||
for i := range vals {
|
||||
ptrs[i] = &vals[i]
|
||||
}
|
||||
// The destination migrator creates a fresh singleton directory revision at
|
||||
// schema startup. Replace it with the source's opaque revision so existing
|
||||
// cursors/visible-change history remain tied to the migrated account data.
|
||||
if table == "account_directory_revision" {
|
||||
if _, err := tx.Exec(`DELETE FROM account_directory_revision`); err != nil {
|
||||
// The destination migrator seeds fresh singletons (directory revision,
|
||||
// authority epoch, role import state) at schema startup. Replace them with
|
||||
// the source's rows so existing cursors/visible-change history and authority
|
||||
// state remain tied to the migrated data.
|
||||
if schemaSeededSingletons[table] {
|
||||
if _, err := tx.Exec(`DELETE FROM "` + table + `"`); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"agent-system-backend/internal/store"
|
||||
)
|
||||
|
||||
// 新库 schema 预置的单行表:必须在 copyOrder 内、确由迁移预置、不计入
|
||||
// businessRows,且复制时由源库行替换(而非 ON CONFLICT 保留目标预置行)。
|
||||
|
||||
func openFreshStore(t *testing.T, name string) *store.Store {
|
||||
t.Helper()
|
||||
st, err := store.OpenSQLite(filepath.Join(t.TempDir(), name))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { st.Close() })
|
||||
return st
|
||||
}
|
||||
|
||||
func tableColumns(t *testing.T, db *sql.DB, table string) []string {
|
||||
t.Helper()
|
||||
rows, err := db.Query(`SELECT name FROM pragma_table_info(?) ORDER BY cid`, table)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var cols []string
|
||||
for rows.Next() {
|
||||
var c string
|
||||
if err := rows.Scan(&c); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cols = append(cols, c)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(cols) == 0 {
|
||||
t.Fatalf("%s 无列", table)
|
||||
}
|
||||
return cols
|
||||
}
|
||||
|
||||
func TestSchemaSeededSingletonsAreSeededAndCopied(t *testing.T) {
|
||||
inOrder := map[string]bool{}
|
||||
for _, tbl := range copyOrder {
|
||||
inOrder[tbl] = true
|
||||
}
|
||||
fresh := openFreshStore(t, "fresh.db").DB()
|
||||
for tbl := range schemaSeededSingletons {
|
||||
if !inOrder[tbl] {
|
||||
t.Errorf("%s 不在 copyOrder 内", tbl)
|
||||
}
|
||||
var n int
|
||||
if err := fresh.QueryRow(`SELECT COUNT(*) FROM "` + tbl + `"`).Scan(&n); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Errorf("%s 新库预置行数=%d,期望 1", tbl, n)
|
||||
}
|
||||
}
|
||||
n, err := businessRows(fresh)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatalf("新库 businessRows=%d,预置单行表不得计为业务数据", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopyReplacesSeededAuthorityState(t *testing.T) {
|
||||
src := openFreshStore(t, "src.db").DB()
|
||||
if _, err := src.Exec(`UPDATE authority_state SET authority_epoch=7, catalog_version=9 WHERE id=1`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dst := openFreshStore(t, "dst.db").DB()
|
||||
cols := tableColumns(t, src, "authority_state")
|
||||
tx, err := dst.Begin()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if err := copyRowsInTx(tx, src, "authority_state", cols, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var count int
|
||||
var epoch, catalog int64
|
||||
if err := dst.QueryRow(`SELECT COUNT(*), MAX(authority_epoch), MAX(catalog_version) FROM authority_state`).Scan(&count, &epoch, &catalog); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 1 || epoch != 7 || catalog != 9 {
|
||||
t.Fatalf("目标 authority_state 未被源行替换:count=%d epoch=%d catalog=%d", count, epoch, catalog)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user