feat: add sync checkpoint restore for last successful sync

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>
This commit is contained in:
whm
2026-08-06 11:49:00 +08:00
parent 75622a728e
commit 680d2b8cde
11 changed files with 860 additions and 16 deletions

View File

@@ -99,6 +99,19 @@ Binding`POST /api/v1/admin/sync/bindings` 登记 `local_database_id → onlin
**空表也要两侧建齐**:本机空表 → `POST .../schema/ensure`;线上空表 → `POST .../schema`(或 pull 的 `columns`)在本机 `CREATE IF NOT EXISTS`。仅靠 outbox 行 push **不会**带上空表。
控制台「查看线上表」可 **删表**`POST .../admin/sync/channels/{id}/drop-table`):只删当前查看侧,**不同步** `DROP` 到另一侧;本机仍有同名表时下次 push/ensure 可能再建回来。
## 数据恢复(恢复上次同步 · 已落实)
成功同步后agent **自动/手动 push**、通道 drain、控制台「同步修复」平台对**线上库**打滚动快照(`latest` + `previous`,约 30 秒防抖)。
| 操作 | 说明 |
|------|------|
| SyncPage「数据恢复」 | 将线上库恢复为所选快照(写入快照行,并删除快照中不存在的行) |
| 本机 | 平台不直连宇恒 SQLite恢复线上后终端下次同步/指纹 pull 可从线上补回 |
| 无快照 | 部署后须先完成至少一次成功同步;按钮会提示「尚无成功同步快照」 |
| 误删后又自动同步成功 | 最新快照可能已含删除后状态 → 确认框选「上一代」 |
API`GET .../admin/sync/channels/{id}/checkpoint``POST .../admin/sync/channels/{id}/restore``confirm=true``which=latest|previous`)。详见联调意见 §5.13。
## 谁能看什么
| 角色 | 可见 |

View File

@@ -74,6 +74,8 @@ var Catalog = []Entry{
{Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/start", OperationID: "startSyncChannel", Summary: "启动近实时同步", Group: "admin"},
{Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/stop", OperationID: "stopSyncChannel", Summary: "停止同步", Group: "admin"},
{Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/reconcile", OperationID: "reconcileSyncChannel", Summary: "主键对账(同步修复,有限流)", Group: "admin"},
{Method: "GET", Path: "/api/v1/admin/sync/channels/{id}/checkpoint", OperationID: "getSyncCheckpoint", Summary: "线上库同步快照摘要latest/previous", Group: "admin"},
{Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/restore", OperationID: "restoreSyncChannel", Summary: "按上次同步快照恢复线上库", Group: "admin"},
{Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/ingest", OperationID: "ingestSyncRows", Summary: "外部行写入通道 local", Group: "admin"},
{Method: "GET", Path: "/api/v1/admin/sync/bindings", OperationID: "listSyncBindings", Summary: "列出本机库↔线上库绑定(用户仅本人)", Group: "admin"},
{Method: "POST", Path: "/api/v1/admin/sync/bindings", OperationID: "ensureSyncBinding", Summary: "登记/更新绑定(用户可自助)", Group: "admin"},

View File

@@ -0,0 +1,453 @@
package dbsync
import (
"compress/gzip"
"context"
"database/sql"
"encoding/json"
"fmt"
"io"
"log"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
const (
checkpointDebounce = 30 * time.Second
checkpointMaxRows = 500_000 // 单次快照行数上限,避免撑爆磁盘
checkpointFileLatest = "latest.json.gz"
checkpointFilePrev = "previous.json.gz"
checkpointFileMeta = "meta.json"
)
// CheckpointSlotMeta 一代快照摘要(不读全量)。
type CheckpointSlotMeta struct {
SyncedAt time.Time `json:"synced_at"`
Source string `json:"source"`
TableCount int `json:"table_count"`
RowCount int `json:"row_count"`
}
// CheckpointMeta 通道快照目录摘要。
type CheckpointMeta struct {
ChannelID string `json:"channel_id"`
Latest *CheckpointSlotMeta `json:"latest,omitempty"`
Previous *CheckpointSlotMeta `json:"previous,omitempty"`
}
// CheckpointTable 单表快照。
type CheckpointTable struct {
PKColumn string `json:"pk_column"`
Rows map[string]map[string]any `json:"rows"` // pk -> row
}
// CheckpointPayload 全量快照内容。
type CheckpointPayload struct {
ChannelID string `json:"channel_id"`
SyncedAt time.Time `json:"synced_at"`
Source string `json:"source"`
Tables map[string]CheckpointTable `json:"tables"`
}
// RestoreResult 恢复结果。
type RestoreResult struct {
OK bool `json:"ok"`
Which string `json:"which"`
SyncedAt time.Time `json:"synced_at"`
Source string `json:"source,omitempty"`
Tables int `json:"tables"`
Upserted int `json:"upserted"`
Deleted int `json:"deleted"`
RestoredAt time.Time `json:"restored_at"`
}
var (
cpSchedMu sync.Mutex
cpPending = map[string]*time.Timer{}
)
func (s *FileStore) Dir() string {
if s == nil {
return "./data/dbsync"
}
return s.dir
}
func checkpointDir(store *FileStore, channelID string) string {
return filepath.Join(store.Dir(), "checkpoints", strings.TrimSpace(channelID))
}
// ScheduleCheckpoint 防抖后写入线上库快照;失败仅打日志。
func ScheduleCheckpoint(store *FileStore, ch *Channel, source string) {
if store == nil || ch == nil || strings.TrimSpace(ch.ID) == "" {
return
}
id := strings.TrimSpace(ch.ID)
chCopy := *ch
src := strings.TrimSpace(source)
if src == "" {
src = "sync"
}
cpSchedMu.Lock()
defer cpSchedMu.Unlock()
if t, ok := cpPending[id]; ok {
t.Stop()
}
cpPending[id] = time.AfterFunc(checkpointDebounce, func() {
cpSchedMu.Lock()
delete(cpPending, id)
cpSchedMu.Unlock()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
if err := SaveCheckpoint(ctx, store, &chCopy, src); err != nil {
log.Printf("dbsync checkpoint channel=%s source=%s: %v", id, src, err)
}
})
}
// SaveCheckpoint 导出通道 Remote 业务表,轮转 previous ← latest。
func SaveCheckpoint(ctx context.Context, store *FileStore, ch *Channel, source string) error {
if store == nil || ch == nil {
return fmt.Errorf("store/channel required")
}
dir := checkpointDir(store, ch.ID)
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
payload, err := dumpRemoteCheckpoint(ctx, ch, source)
if err != nil {
return err
}
latestPath := filepath.Join(dir, checkpointFileLatest)
prevPath := filepath.Join(dir, checkpointFilePrev)
// 轮转:现有 latest → previous
if _, err := os.Stat(latestPath); err == nil {
_ = os.Remove(prevPath)
if err := os.Rename(latestPath, prevPath); err != nil {
// Windows 上目标存在时 Rename 可能失败;已删 prev 再试
_ = os.Remove(prevPath)
if err2 := os.Rename(latestPath, prevPath); err2 != nil {
return fmt.Errorf("rotate checkpoint: %w", err2)
}
}
}
tmp := latestPath + ".tmp"
if err := writeCheckpointGzip(tmp, payload); err != nil {
_ = os.Remove(tmp)
return err
}
if err := os.Rename(tmp, latestPath); err != nil {
_ = os.Remove(latestPath)
if err2 := os.Rename(tmp, latestPath); err2 != nil {
_ = os.Remove(tmp)
return err2
}
}
meta := CheckpointMeta{ChannelID: ch.ID}
meta.Latest = slotMetaFromPayload(payload)
if prev, err := loadCheckpointPayload(prevPath); err == nil && prev != nil {
meta.Previous = slotMetaFromPayload(prev)
}
return writeCheckpointMeta(dir, meta)
}
func slotMetaFromPayload(p *CheckpointPayload) *CheckpointSlotMeta {
if p == nil {
return nil
}
rows := 0
for _, t := range p.Tables {
rows += len(t.Rows)
}
return &CheckpointSlotMeta{
SyncedAt: p.SyncedAt,
Source: p.Source,
TableCount: len(p.Tables),
RowCount: rows,
}
}
func dumpRemoteCheckpoint(ctx context.Context, ch *Channel, source string) (*CheckpointPayload, error) {
db, err := AcquireRemote(ch.Remote.Driver, ch.Remote.DSN)
if err != nil {
return nil, wrapOpenRemote(err)
}
names, err := listTablesForInspect(ctx, db, ch.Remote.Driver, false)
if err != nil {
return nil, err
}
out := &CheckpointPayload{
ChannelID: ch.ID,
SyncedAt: time.Now().UTC(),
Source: source,
Tables: make(map[string]CheckpointTable, len(names)),
}
totalRows := 0
for _, table := range names {
pkCol := "id"
if ch.PKColumns != nil && strings.TrimSpace(ch.PKColumns[table]) != "" {
pkCol = strings.TrimSpace(ch.PKColumns[table])
}
cols, err := listColumns(ctx, db, ch.Remote.Driver, table)
if err != nil {
return nil, fmt.Errorf("columns %s: %w", table, err)
}
rows, err := fetchAllRows(ctx, db, ch.Remote.Driver, table, cols)
if err != nil {
return nil, fmt.Errorf("dump %s: %w", table, err)
}
m := make(map[string]map[string]any, len(rows))
for _, row := range rows {
pk := fmt.Sprint(row[pkCol])
if pk == "" || pk == "<nil>" {
continue
}
m[pk] = row
}
totalRows += len(m)
if totalRows > checkpointMaxRows {
return nil, fmt.Errorf("checkpoint too large: >%d rows", checkpointMaxRows)
}
out.Tables[table] = CheckpointTable{PKColumn: pkCol, Rows: m}
}
return out, nil
}
func fetchAllRows(ctx context.Context, db *sql.DB, driver Driver, table string, cols []string) ([]map[string]any, error) {
if len(cols) == 0 {
return nil, nil
}
q := fmt.Sprintf(`SELECT * FROM %s`, quoteIdent(driver, table))
rows, err := db.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
colNames, err := rows.Columns()
if err != nil {
return nil, err
}
var out []map[string]any
for rows.Next() {
raw := make([]any, len(colNames))
ptrs := make([]any, len(colNames))
for i := range raw {
ptrs[i] = &raw[i]
}
if err := rows.Scan(ptrs...); err != nil {
return nil, err
}
m := make(map[string]any, len(colNames))
for i, c := range colNames {
m[c] = normalizeValue(raw[i])
}
out = append(out, m)
}
return out, rows.Err()
}
func writeCheckpointGzip(path string, payload *CheckpointPayload) error {
f, err := os.Create(path)
if err != nil {
return err
}
defer f.Close()
zw := gzip.NewWriter(f)
enc := json.NewEncoder(zw)
if err := enc.Encode(payload); err != nil {
_ = zw.Close()
return err
}
if err := zw.Close(); err != nil {
return err
}
return f.Close()
}
func loadCheckpointPayload(path string) (*CheckpointPayload, error) {
f, err := os.Open(path)
if err != nil {
return nil, err
}
defer f.Close()
zr, err := gzip.NewReader(f)
if err != nil {
return nil, err
}
defer zr.Close()
var p CheckpointPayload
if err := json.NewDecoder(zr).Decode(&p); err != nil && err != io.EOF {
return nil, err
}
return &p, nil
}
func writeCheckpointMeta(dir string, meta CheckpointMeta) error {
b, err := json.MarshalIndent(meta, "", " ")
if err != nil {
return err
}
return os.WriteFile(filepath.Join(dir, checkpointFileMeta), b, 0o644)
}
// LoadCheckpointMeta 读取摘要;无快照时返回空 meta不报错
func LoadCheckpointMeta(store *FileStore, channelID string) (*CheckpointMeta, error) {
if store == nil || strings.TrimSpace(channelID) == "" {
return &CheckpointMeta{}, nil
}
dir := checkpointDir(store, channelID)
b, err := os.ReadFile(filepath.Join(dir, checkpointFileMeta))
if err != nil {
if os.IsNotExist(err) {
// 尝试从文件推断
meta := &CheckpointMeta{ChannelID: channelID}
if p, e := loadCheckpointPayload(filepath.Join(dir, checkpointFileLatest)); e == nil {
meta.Latest = slotMetaFromPayload(p)
}
if p, e := loadCheckpointPayload(filepath.Join(dir, checkpointFilePrev)); e == nil {
meta.Previous = slotMetaFromPayload(p)
}
return meta, nil
}
return nil, err
}
var meta CheckpointMeta
if err := json.Unmarshal(b, &meta); err != nil {
return nil, err
}
meta.ChannelID = channelID
return &meta, nil
}
// LoadCheckpoint 加载 latest 或 previous。
func LoadCheckpoint(store *FileStore, channelID, which string) (*CheckpointPayload, error) {
which = strings.TrimSpace(which)
if which == "" {
which = "latest"
}
if which != "latest" && which != "previous" {
return nil, fmt.Errorf("which 须为 latest 或 previous")
}
name := checkpointFileLatest
if which == "previous" {
name = checkpointFilePrev
}
path := filepath.Join(checkpointDir(store, channelID), name)
p, err := loadCheckpointPayload(path)
if err != nil {
if os.IsNotExist(err) {
return nil, fmt.Errorf("无可用快照(%s", which)
}
return nil, err
}
return p, nil
}
// RestoreCheckpoint 将线上库恢复为指定快照upsert + 删除快照外 PK
func RestoreCheckpoint(ctx context.Context, store *FileStore, ch *Channel, which string) (*RestoreResult, error) {
if store == nil || ch == nil {
return nil, fmt.Errorf("store/channel required")
}
payload, err := LoadCheckpoint(store, ch.ID, which)
if err != nil {
return nil, err
}
db, err := AcquireRemote(ch.Remote.Driver, ch.Remote.DSN)
if err != nil {
return nil, wrapOpenRemote(err)
}
if err := EnsureMeta(ctx, db, ch.Remote.Driver); err != nil {
return nil, err
}
verBase := time.Now().UnixNano()
upserted, deleted := 0, 0
for table, ct := range payload.Tables {
pkCol := strings.TrimSpace(ct.PKColumn)
if pkCol == "" {
pkCol = "id"
if ch.PKColumns != nil && strings.TrimSpace(ch.PKColumns[table]) != "" {
pkCol = strings.TrimSpace(ch.PKColumns[table])
}
}
// 先 upsert 快照行
i := 0
for pk, row := range ct.Rows {
if len(row) == 0 {
continue
}
if err := EnsureTableFromRow(ctx, db, ch.Remote.Driver, table, pkCol, row); err != nil {
return nil, fmt.Errorf("ensure %s: %w", table, err)
}
b, err := json.Marshal(row)
if err != nil {
return nil, err
}
ver := verBase + int64(i)
i++
if err := ApplyChange(ctx, db, ch.Remote.Driver, table, pkCol, "upsert", string(b), ver); err != nil {
return nil, fmt.Errorf("upsert %s pk=%s: %w", table, pk, err)
}
upserted++
}
// 删快照中不存在的行(含空表:清空线上多余行)
livePKs, err := listTablePKs(ctx, db, ch.Remote.Driver, table, pkCol)
if err != nil {
// 表可能尚不存在且快照也空
if len(ct.Rows) == 0 {
continue
}
return nil, fmt.Errorf("list pk %s: %w", table, err)
}
for _, pk := range livePKs {
if _, ok := ct.Rows[pk]; ok {
continue
}
delPayload, _ := json.Marshal(map[string]any{pkCol: pk})
ver := verBase + int64(i)
i++
if err := ApplyChange(ctx, db, ch.Remote.Driver, table, pkCol, "delete", string(delPayload), ver); err != nil {
return nil, fmt.Errorf("delete %s pk=%s: %w", table, pk, err)
}
deleted++
}
}
// 快照里没有、线上多出来的业务表:不自动 DROP避免误伤仅对快照内表做行级 prune
which = strings.TrimSpace(which)
if which == "" {
which = "latest"
}
return &RestoreResult{
OK: true,
Which: which,
SyncedAt: payload.SyncedAt,
Source: payload.Source,
Tables: len(payload.Tables),
Upserted: upserted,
Deleted: deleted,
RestoredAt: time.Now().UTC(),
}, nil
}
func listTablePKs(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string) ([]string, error) {
q := fmt.Sprintf(`SELECT %s FROM %s`, quoteIdent(driver, pkCol), quoteIdent(driver, table))
rows, err := db.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
var out []string
for rows.Next() {
var v any
if err := rows.Scan(&v); err != nil {
return nil, err
}
s := fmt.Sprint(normalizeValue(v))
if s != "" && s != "<nil>" {
out = append(out, s)
}
}
return out, rows.Err()
}

View File

@@ -0,0 +1,89 @@
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)
}
}

View File

@@ -120,6 +120,7 @@ func (m *Manager) loop(ctx context.Context, id string) {
now := time.Now().UTC()
c.LastReconcileAt = &now
})
ScheduleCheckpoint(m.store, ch, "reconcile")
}
}
}
@@ -176,6 +177,10 @@ func (m *Manager) tick(ctx context.Context, ch *Channel) error {
c.LastError = ""
}
})
// 成功且本批有变更 → 防抖打线上库快照(数据恢复)
if err == nil && n > 0 {
ScheduleCheckpoint(m.store, ch, "drain")
}
return err
}

View File

@@ -90,6 +90,9 @@ func agentSyncPushHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
if res != nil && res.Skipped {
outcome = "skipped"
}
if res != nil && res.Applied {
dbsync.ScheduleCheckpoint(svcCtx.DBSync.Store(), ch, "push")
}
logSyncReq(r, "push", channelID, item.Table, item.RowPK, outcome, "", dur, reqID)
auditSync(svcCtx, r, "dbsync.push", map[string]any{
"channel_id": channelID, "table": item.Table, "row_pk": item.RowPK,
@@ -153,6 +156,15 @@ func agentSyncPushBatchHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
httpx.OkJson(w, payload)
return
}
applied := 0
for _, r0 := range results {
if r0.Applied {
applied++
}
}
if applied > 0 {
dbsync.ScheduleCheckpoint(svcCtx.DBSync.Store(), ch, "push_batch")
}
logSyncReq(r, "push_batch", channelID, "", "", "ok", "", dur, reqID)
auditSync(svcCtx, r, "dbsync.push_batch", map[string]any{
"channel_id": channelID, "n": len(body.Items), "ms": dur.Milliseconds(), "req_id": reqID,

View File

@@ -146,6 +146,8 @@ func RegisterHandlers(server *rest.Server, svcCtx *svc.ServiceContext) {
{Method: http.MethodGet, Path: "/api/v1/admin/sync/conflicts", Handler: chain(syncConflictsHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodPost, Path: "/api/v1/admin/sync/conflicts/:id/resolve", Handler: chain(syncResolveConflictHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/reconcile", Handler: chain(syncReconcileHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodGet, Path: "/api/v1/admin/sync/channels/:id/checkpoint", Handler: chain(syncCheckpointMetaHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/restore", Handler: chain(syncRestoreHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/ingest", Handler: chain(syncIngestHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodGet, Path: "/api/v1/admin/sync/channels/:id/inspect", Handler: chain(syncInspectHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},
{Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/preview", Handler: chain(syncPreviewHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))},

View File

@@ -262,6 +262,7 @@ func syncReconcileHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
_ = svcCtx.DBSync.Store().PatchStats(ch.ID, func(c *dbsync.Channel) {
c.LastReconcileAt = &now
})
dbsync.ScheduleCheckpoint(svcCtx.DBSync.Store(), ch, "reconcile")
httpx.OkJson(w, res)
}
}
@@ -418,3 +419,62 @@ func syncDropTableHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
httpx.OkJson(w, res)
}
}
func syncCheckpointMetaHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if !requireDBSync(svcCtx, w) {
return
}
id := pathvar.Vars(r)["id"]
if _, err := svcCtx.DBSync.Store().GetChannelForTenant(id, syncTenantID(r)); err != nil {
authx.WriteError(w, http.StatusNotFound, err.Error())
return
}
meta, err := dbsync.LoadCheckpointMeta(svcCtx.DBSync.Store(), id)
if err != nil {
authx.WriteError(w, http.StatusBadRequest, err.Error())
return
}
httpx.OkJson(w, meta)
}
}
func syncRestoreHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if !requireDBSync(svcCtx, w) {
return
}
ch, err := svcCtx.DBSync.Store().GetChannelForTenant(pathvar.Vars(r)["id"], syncTenantID(r))
if err != nil {
authx.WriteError(w, http.StatusNotFound, err.Error())
return
}
var body struct {
Which string `json:"which"`
Confirm bool `json:"confirm"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
authx.WriteError(w, http.StatusBadRequest, err.Error())
return
}
if !body.Confirm {
authx.WriteError(w, http.StatusBadRequest, "请确认恢复confirm=true")
return
}
which := strings.TrimSpace(body.Which)
if which == "" {
which = "latest"
}
res, err := dbsync.RestoreCheckpoint(r.Context(), svcCtx.DBSync.Store(), ch, which)
if err != nil {
authx.WriteError(w, http.StatusBadRequest, err.Error())
return
}
if svcCtx.Audit != nil {
_ = svcCtx.Audit.Log(r.Context(), syncTenantID(r), authx.UserID(r.Context()), "dbsync.restore", audit.DetailJSON(map[string]any{
"channel_id": ch.ID, "which": which, "synced_at": res.SyncedAt, "upserted": res.Upserted, "deleted": res.Deleted,
}))
}
httpx.OkJson(w, res)
}
}

View File

@@ -34,6 +34,7 @@ import {
createSyncChannel,
deleteSyncChannel,
dropSyncTable,
getSyncCheckpointMeta,
inspectSyncChannel,
listAgents,
listApps,
@@ -44,6 +45,7 @@ import {
revokeBindCode,
previewSyncTable,
reconcileSyncChannel,
restoreSyncChannel,
startSyncChannel,
stopSyncChannel,
testSyncEndpoints,
@@ -308,6 +310,83 @@ export function SyncPage(props: {
}
}
function formatCheckpointTime(iso?: string) {
if (!iso) return "—";
const d = new Date(iso);
if (Number.isNaN(d.getTime())) return iso;
return d.toLocaleString();
}
async function openRestore(r: SyncChannel) {
setBusy(true);
try {
const meta = await getSyncCheckpointMeta(session, r.id);
if (!meta.latest && !meta.previous) {
message.warning("尚无成功同步快照。请先完成一次同步(含自动推送),约 30 秒后生成。");
return;
}
let which: "latest" | "previous" = meta.latest ? "latest" : "previous";
const options: { value: "latest" | "previous"; label: string }[] = [];
if (meta.latest) {
options.push({
value: "latest",
label: `最近一次 · ${formatCheckpointTime(meta.latest.synced_at)}${meta.latest.row_count ?? 0} 行 / ${meta.latest.table_count ?? 0} 表)`,
});
}
if (meta.previous) {
options.push({
value: "previous",
label: `上一代 · ${formatCheckpointTime(meta.previous.synced_at)}${meta.previous.row_count ?? 0} 行 / ${meta.previous.table_count ?? 0} 表)`,
});
}
Modal.confirm({
title: `数据恢复 · ${r.name || r.id}`,
width: 560,
content: (
<div>
<p style={{ marginTop: 0 }}>
<strong>线</strong>upsert /pull 线
</p>
<p style={{ color: "rgba(0,0,0,0.45)", fontSize: 13 }}>
</p>
<div style={{ marginTop: 12 }}>
<Typography.Text type="secondary"></Typography.Text>
<Select
style={{ width: "100%", marginTop: 6 }}
defaultValue={which}
options={options}
onChange={(v) => {
which = v;
}}
/>
</div>
</div>
),
okText: "确认恢复",
okType: "danger",
cancelText: "取消",
onOk: async () => {
try {
const res = await restoreSyncChannel(session, r.id, { which, confirm: true });
message.success(
`已恢复:写入 ${res.upserted ?? 0} 行,清理多余 ${res.deleted ?? 0} 行(快照 ${formatCheckpointTime(res.synced_at)}`
);
setInfo("线上库已按同步快照恢复");
await refresh();
} catch (e: any) {
message.error(e.message || String(e));
throw e;
}
},
});
} catch (e: any) {
message.error(e.message || String(e));
} finally {
setBusy(false);
}
}
function confirmDropTable(table: string) {
if (!inspectCh) return;
const sideLabel = inspectSide === "remote" ? "线上库" : "本机端(通道 local";
@@ -583,6 +662,9 @@ export function SyncPage(props: {
>
</Button>
<Button size="small" onClick={() => void openRestore(r)}>
</Button>
<Button
size="small"
danger

View File

@@ -1063,6 +1063,55 @@ export async function reconcileSyncChannel(session: Session, id: string) {
};
}
export type SyncCheckpointSlot = {
synced_at?: string;
source?: string;
table_count?: number;
row_count?: number;
};
export type SyncCheckpointMeta = {
channel_id?: string;
latest?: SyncCheckpointSlot | null;
previous?: SyncCheckpointSlot | null;
};
export async function getSyncCheckpointMeta(session: Session, id: string) {
const res = await apiFetch(`/api/v1/admin/sync/channels/${encodeURIComponent(id)}/checkpoint`, {
headers: { Authorization: `Bearer ${session.accessToken}` },
});
const data = await readJson(res);
throwIfBad(res, data, "get checkpoint failed");
return data as SyncCheckpointMeta;
}
export async function restoreSyncChannel(
session: Session,
id: string,
body: { which?: "latest" | "previous"; confirm: boolean }
) {
const res = await apiFetch(`/api/v1/admin/sync/channels/${encodeURIComponent(id)}/restore`, {
method: "POST",
headers: {
Authorization: `Bearer ${session.accessToken}`,
"Content-Type": "application/json",
},
body: JSON.stringify(body),
});
const data = await readJson(res);
throwIfBad(res, data, "restore failed");
return data as {
ok?: boolean;
which?: string;
synced_at?: string;
source?: string;
tables?: number;
upserted?: number;
deleted?: number;
restored_at?: string;
};
}
/** 外部源 C → 写入本地 B再经 outbox 同步到线上 A */
export async function ingestSyncRows(
session: Session,

View File

@@ -1,7 +1,7 @@
# 联调后修改意见 · 宇恒松离线(形态 B
> 初稿2026-08-01 · 修订至 **2026-08-06**Z12h 已落实)
> 焦点:**§0.2**(按负责方:宇恒改宇恒、智建改智建);配合见 **§0.3**
> 初稿2026-08-01 · 修订至 **2026-08-06**Z12h + 数据恢复已落实)
> 焦点:**§0.2**(按负责方**改代码前须先写入本意见**(见 §0.0
> 来源:宇恒 `yuhengyihao_client` ↔ 智建(生产 `aisite.yuxindazhineng.com`
> 依据:`松离线-dbsync方案-最终版.md`、`宇恒-松离线数据同步使用文档.md`
@@ -16,9 +16,17 @@
| **Z1Z9 / Z11Z13** | **已落实** | 绑定/凭票/模块导入等;见 §5 |
| **Z10 + Z10d** | **代码齐;生产 pull** | 空表 ensure已存在表 **ADD COLUMN**§5.7 |
| **Z12 默认同步** | **已落实(含 Z12h** | 启用即通道;误删 → Ensure+重挂§5.9 |
| **数据恢复** | **智建已落实** | SyncPage「数据恢复」同步成功后打线上快照可恢复上次/上一代§5.13 |
| **Z14 + Z14c** | **代码齐;生产 pull** | fingerprint智能体挂载 `online_db_id`§5.11 |
| **Z14d 双向补齐** | **宇恒已接** | 误删任一侧 → 心跳/选同步按主键并集 pull↔push§5.11 |
| **仍关注** | 用法 | 通道断了靠智建自愈;表数据靠宇恒双向指纹;不是「再点启动」 |
| **Z15 建站配置/绑定优先** | **宇恒已改完;智建无需** | 新机缺 `config.json` 自动生成无权限先绑定再鉴权§5.12 |
| **仍关注** | 用法 | 通道断了靠智建自愈;表数据靠宇恒双向指纹 / 数据恢复;不是「再点启动」 |
### 0.0 修改流程(冻结)
1. **先写本意见**§0.2 分 A/B″§5 写清诉求/分工/验收)→ 双方对齐。
2. **再改代码**:宇恒只改 `yuhengyihao_client`;智建只改 `ai建站`**互不代改**。
3. 接口/产品有变 → 回写 §1 契约与验收;禁止口头改完再补意见。
### 0.1 绑定产品冻结Z13 · 试运行)
@@ -43,6 +51,7 @@
| **Z13be** | 绑定码 / 同号确认 / policy / 凭票 / 选「同步」即 Binding+drain |
| **Z14a** | 表级指纹慢心跳(原:不一致 → ensure + full-push |
| **Z14d** | **双向并集补齐2026-08-06**:不一致 → 先 `pull` 灌本机(不进 outbox再 ensure+full-push选「同步」同样先 pull 再 push**默认不做 prune**`YXD_SYNC_PRUNE_EXTRAS=1` 才清线上多余)。覆盖:线上误删 / 本地误删 / 缺表 |
| **Z15** | **建站技能 config 自生成 + 无权限先绑定2026-08-06 · 宇恒已改完)**`ai-site-agent-api``config.json` 自动写默认模板;绑定优先;门闸含「绑定账号或手机号」。改动:`config_loader.py` / `host_bootstrap.py` / `forms.py` / `forms_ui.py` / `forms_ui.json` / `reference.md`。**智建无需改** |
#### A. 宇恒 · 配合注意(非阻塞新开发)
@@ -63,6 +72,7 @@
| **Z14c** ACL | 智能体 JWT 放行挂载 `online_db_id``agentOwnsOnlineDB` |
| Podman DNS / 脚本可执行位 | compose aliases`restart.sh` 100755 |
| **Z12h / Z12h-1 / Z12h-2** | `HealTenantSyncBind`Ensure 默认通道 + 重挂 Agent/Binding默认同步**可删**,删后立即 Ensure+heal列表/换票/`agents/me`/ensure Binding 触发;见 §5.9 |
| **数据恢复§5.13** | 成功同步后线上库快照latest+previousSyncPage「数据恢复」`GET …/checkpoint` + `POST …/restore` |
> 产品一句:**账号已绑定 ⇒ 落点信息固定;通道没了平台自动补,终端无需手填 channel_id。**
@@ -70,13 +80,23 @@
| 优先级 | 编号 | 项 | 说明 |
|--------|------|----|------|
| — | — | **无阻塞开发项** | Z12h 已合入;后续仅生产 pull 与运维配号 |
| — | — | **无阻塞开发项** | Z12h、数据恢复已合入;后续仅生产 pull 与运维配号 |
#### B‴. 智建 · 本次明确不改Z15
| 项 | 说明 |
|----|------|
| 新机缺 `config.json` | **宇恒技能本地文件**问题;平台不负责生成终端 config |
| 「模块操作 / 等待授权」表单 | 宇恒 `sendFrom` + `yxd_skill_ty_host`;智建已有 bind/register/token API**契约够用** |
| pending 待管理员启用 | 仍靠控制台赋「生成发布」;**不**为此新开接口 |
> 结论:**Z15 智建无需开发**;仅当后续要把「绑定失败错误码 / pending 文案」标准化进 API 时再开智建项并先改本意见。
#### B. 智建 · 仍待办(仅运维)
| 优先级 | 项 | 说明 |
|--------|----|------|
| **P0** | **生产 pull 本批** | Z10d + `schema/fingerprint` + Z14c + **Z12h**`bash ./restart.sh --pull` |
| **P0** | **生产 pull 本批** | Z10d + fingerprint + Z14c + **Z12h** + **数据恢复**`bash ./restart.sh --pull` |
| **P0** | 成员手机 | 「宇信达」绑 **`13531041944`**;勿超管号 |
| **P0** | 通道表白名单 | **勿**「填入测试默认」;形态 B 可空 |
| **P1** | 凭票 Secret | `YuhengTicket.Secret``YXD_YUHENG_TICKET_SECRET` |
@@ -101,8 +121,9 @@
3. 启用智能体 → 默认同步通道应「运行中」
4.(可选)生成绑定码
【智建 · Z12h 已落实】
误删默认同步 / Binding 挂死 → 打开 SyncPage / 换票 / agents/me 自动 Ensure+重挂
【智建 · Z12h / 数据恢复 已落实】
误删默认同步 / Binding 挂死 → SyncPage / 换票 / agents/me 自动 Ensure+重挂
误删线上数据 → SyncPage「数据恢复」按上次同步快照恢复线上本机靠下次 pull
【宇恒终端】
5. 未绑定 → policy → lookup/confirm 或 redeem 或 ticket-exchange
@@ -113,8 +134,8 @@
【智建验收】
9. 「查看线上表」刷新 → 应有业务表(含曾 0 行空表)
10. 仍空 = 终端尚未 push/ensure/pull不是再点一次「启动」就能灌表
11. 故意删默认同步通道后:换票或打开 SyncPage / agents/me → 通道自动回来Binding 不再红字
```
11. 删默认同步后:换票或打开 SyncPage / agents/me → 通道自动回来
12. 同步成功后误删线上行 →「数据恢复」可选最近一次/上一代快照恢复
| 角色 | 负责 | 不负责 |
|------|------|--------|
@@ -551,14 +572,14 @@ POST .../schema
| **Z12e** | **P1** | **唯一默认通道** | 每公司至多 1 条 `is_system_default`;复用已有默认 | **智建已落实** + 宇恒「仅 1 条则写入 env」 |
| **Z12f** | **P1** | **开通 UX** | 启用智能体即可DSN 用公司级默认 | **智建已落实** |
| **Z12g** | **P1** | **默认同步通道默认启动** | 创建 `Enabled=true`;重启对 `IsSystemDefault` 自动 Start控制台保存后自动启动 | **智建已落实**`cb76824` |
| **Z12h** | **P0** | **通道误删自愈** | `HealTenantSyncBind`Ensure 默认通道 + 重挂挂死 Agent/Binding(保留 `online_db_id`);触发:`ensureAgentSyncBind` / `agents/me` / 换票 / 通道·Binding 列表 / 删非默认通道后 | **智建已落实** |
| **Z12h** | **P0** | **通道误删自愈** | `HealTenantSyncBind`Ensure 默认通道 + 重挂挂死 Agent/Binding;触发:列表/换票/`agents/me`/ensure Binding/删通道后 | **智建已落实** |
| **Z12h-1** | **P1** | **禁删或删后重建默认同步** | `is_system_default` **可删**;删后立即 Ensure+heal 重建并重挂 | **智建已落实** |
| **Z12h-2** | **P1** | **orphan Binding heal** | 列表 / Ensure Binding 时通道缺失 → 静默改挂默认通道 | **智建已落实** |
**缺口说明2026-08-06 · 已关闭)**
-SyncPage「通道已删除」、`ensureAgentSyncBind` 见非空 `channel_id` 即跳过。
- 现:列表/换票等路径先 heal默认通道可删并立即重建。生产须 **pull** 后验收第 4 条
- 现:列表/换票等路径先 heal默认同步可删并立即重建。生产须 **pull** 后验收。
**宇恒已做2026-08-05**
@@ -571,7 +592,7 @@ POST .../schema
1. 新公司:只「启用智能体」,**不**点新建通道,换票已带 `channel_id`,宇恒可 drain。
2. 账号 A 的本机库 push 不会出现在账号 B 的线上库;共享库仅在显式共享时可见。
3. 用户全程无需 F12、无需手填 `YXD_SYNC_CHANNEL_ID`
4. **Z12h** 删除默认同步通道 → 立即重建并重挂Binding/Agent 挂死 ID 时打开 SyncPage / 换票 / `agents/me` 亦会自愈,「通道已删除」红字消失;宇恒无需手填新 ID
4. **Z12h** 删除默认同步通道 → 立即重建并重挂Binding 挂死 ID 时打开 SyncPage / 换票亦自愈
### 5.10 【双方代码已接 Z13】绑定流程简化绑定码 / 手机号2026-08-05
@@ -684,7 +705,7 @@ POST /api/v1/agent/sync/channels/{id}/pull
| 方 | 做什么 |
|----|--------|
| 宇恒 | 本机指纹404 回退 `row_count`**Z14d** 双向 pull↔push选同步默认不 prune |
| 智建 | 生产 pullfingerprint + Z10d + Z14c + **Z12h**;保持 pull/bootstrap 契约 |
| 智建 | 生产 pullfingerprint + Z10d + Z14c + **Z12h** + **数据恢复**;保持 pull/bootstrap 契约 |
**联调踩坑(宇信达 · 已回写)**
@@ -702,14 +723,70 @@ POST /api/v1/agent/sync/channels/{id}/pull
5. **Z14d** 线上删某表行 → 心跳后本机行仍在且线上恢复;本机删某表行(未 prune→ 心跳后本机从线上拉回。
6. **Z12h** 见 §5.9 验收第 4 条。
### 5.12 【Z15 · 2026-08-06】建站技能 config 自生成 + 无权限先绑定
**状态****宇恒已改完2026-08-06****智建不改 / 无需开发**。
落地文件(宇恒仓 `.yxd/skills/user/ai-site-agent-api/``scripts/config_loader.py``ensure_config`)、`scripts/host_bootstrap.py`(绑定优先)、`scripts/forms.py``scripts/forms_ui.py``forms_ui.json``reference.md`
**现象(另一台电脑)**
- 克隆后无 `config.json`gitignore 密钥文件)→ `默认登记失败: [Errno 2] No such file or directory: '.../config.json'`
- 随后卡在鉴权/换票,而不是先走绑定码/手机号
**诉求**
1.`config.json` → 自动写默认模板(可再填绑定码/手机号等),**禁止**裸抛 Errno 2
2. 启动顺序:**绑定优先** → 再鉴权/换票;未绑定或无权限时门闸选项含「绑定账号或手机号」
3. 聊天表单仍走宇恒 `sendFrom` / `yxd_skill_ty_host`(与税务 TY 插件无关,但是同一套交互壳)
**分工**
| 方 | 做什么 |
|----|--------|
| 宇恒 | `ai-site-agent-api``config_loader.ensure_config``host_bootstrap` 绑定优先、`forms`/`forms_ui` 门闸选项 |
| 智建 | **不改**bind / register / ticket 已够用(见 §0.2 B‴ |
**宇恒落地(已改完)**
1. `ensure_config()`:缺 `config.json` 写默认模板,不再 Errno 2
2. `host_bootstrap`:绑定码/手机号优先于鉴权换票
3. `pending_auth` 等门闸选项含「绑定账号或手机号」
**验收**
1. 新机无 config → 首次跑技能生成模板文件,无 Errno 2
2. 未绑定 → 弹绑定相关表单,而非只提示换票失败
3. 已绑定有权限 → 原鉴权/模块操作路径不变
### 5.13 【智建已落实 · 2026-08-06】数据恢复恢复上次同步
**状态****智建已落实**(须生产 `bash ./restart.sh --pull`)。
**诉求**10:00 同步成功 → 11:00 误删 → 可恢复到最后一次成功同步(含自动 push/drain时的线上库状态。
| 项 | 说明 |
|----|------|
| 快照 | 成功同步后对通道 **Remote 线上库** 打滚动快照(`latest` + `previous`,约 30s 防抖) |
| 触发 | agent push/batch有 applied、通道 drain 有变更、控制台「同步修复」成功 |
| UI | SyncPage 通道操作「**数据恢复**」;可选最近一次 / 上一代 |
| API | `GET /api/v1/admin/sync/channels/{id}/checkpoint``POST …/restore``confirm=true``which=latest\|previous` |
| 恢复范围 | **仅线上库**;本机靠宇恒下次同步/指纹 pull 从线上补回 |
| 注意 | 误删后若又自动同步成功,最新快照可能已含删除态 → 选「上一代」或尽快恢复 |
**验收**
1. 同步成功约 30s 后checkpoint 有 `latest`
2. 删线上若干行 →「数据恢复」选最近一次 → 「查看线上表」行数/内容回到快照。
3. 无快照时按钮提示「尚无成功同步快照」,不静默失败。
---
## 6. 联系与附件
- **待改清单(优先看)**§0.2**A 宇恒配合 · B 智建运维**;开发项已清空);配合见 **§0.3**
- **待改清单(优先看)**§0.2**A 宇恒配合 · B 智建运维**;开发项已清空);**改代码前先写本意见**§0.0配合见 **§0.3**
- 方案:`松离线-dbsync方案-最终版.md`(含 2026-08-01 联调建议落地记录)
- 宇恒使用说明:`宇恒-松离线数据同步使用文档.md`(含 Z10 schema/ensure、Z13 绑定)
- 开通说明:`docs/数据同步-开通说明.md`
- 开通说明:`docs/数据同步-开通说明.md`(含 **数据恢复**
- 绑定策略探测:`GET /api/v1/auth/bind/policy``trial_mode` / `require_for_bind`
- 本意见如与冻结方案冲突,**以冻结方案为准**§5.15.11 为产品增量与复测记录,不推翻 H1H6 默认无感约束。
- 本意见如与冻结方案冲突,**以冻结方案为准**§5.15.13 为产品增量与复测记录,不推翻 H1H6 默认无感约束。
- **分工**:宇恒只改宇恒仓;智建只改智建仓;互不代改。