whm
680d2b8cde
Capture rolling online DB snapshots after push/drain/reconcile and expose SyncPage 数据恢复 plus checkpoint/restore APIs; mark Z12h and restore done in coop docs. Co-authored-by: Cursor <cursoragent@cursor.com>
90 lines
2.4 KiB
Go
90 lines
2.4 KiB
Go
package dbsync
|
|
|
|
import (
|
|
"context"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestCheckpointSaveRestoreRoundTrip(t *testing.T) {
|
|
dir := t.TempDir()
|
|
store, err := NewFileStore(dir)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
remoteDSN := "file:" + filepath.ToSlash(filepath.Join(dir, "online.db")) + "?_pragma=busy_timeout(5000)"
|
|
ch := &Channel{
|
|
ID: "ch-cp-1",
|
|
TenantID: 1,
|
|
Name: "test",
|
|
Remote: Endpoint{
|
|
Driver: DriverSQLite,
|
|
DSN: remoteDSN,
|
|
},
|
|
PKColumns: map[string]string{"demo": "id"},
|
|
}
|
|
ctx := context.Background()
|
|
db, err := Open(DriverSQLite, remoteDSN)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() {
|
|
_ = db.Close()
|
|
InvalidateRemote(DriverSQLite, remoteDSN)
|
|
})
|
|
if _, err := db.ExecContext(ctx, `CREATE TABLE demo (id TEXT PRIMARY KEY, name TEXT)`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := db.ExecContext(ctx, `INSERT INTO demo(id,name) VALUES('a','one'),('b','two')`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := EnsureMeta(ctx, db, DriverSQLite); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if err := SaveCheckpoint(ctx, store, ch, "test"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
meta, err := LoadCheckpointMeta(store, ch.ID)
|
|
if err != nil || meta.Latest == nil || meta.Latest.RowCount != 2 {
|
|
t.Fatalf("meta=%+v err=%v", meta, err)
|
|
}
|
|
|
|
// mutate then restore
|
|
if _, err := db.ExecContext(ctx, `DELETE FROM demo WHERE id='a'`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := db.ExecContext(ctx, `INSERT INTO demo(id,name) VALUES('c','three')`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
res, err := RestoreCheckpoint(ctx, store, ch, "latest")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !res.OK || res.Upserted < 2 {
|
|
t.Fatalf("restore=%+v", res)
|
|
}
|
|
var n int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM demo`).Scan(&n); err != nil || n != 2 {
|
|
t.Fatalf("count=%d err=%v", n, err)
|
|
}
|
|
var name string
|
|
if err := db.QueryRowContext(ctx, `SELECT name FROM demo WHERE id='a'`).Scan(&name); err != nil || name != "one" {
|
|
t.Fatalf("row a name=%q err=%v", name, err)
|
|
}
|
|
|
|
// second save rotates
|
|
time.Sleep(10 * time.Millisecond)
|
|
if _, err := db.ExecContext(ctx, `UPDATE demo SET name='one2' WHERE id='a'`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := SaveCheckpoint(ctx, store, ch, "test2"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
meta2, _ := LoadCheckpointMeta(store, ch.ID)
|
|
if meta2.Previous == nil || meta2.Previous.RowCount != 2 {
|
|
t.Fatalf("expected previous after rotate: %+v", meta2)
|
|
}
|
|
}
|