fix(persist,channel): 同类定宽时间戳修复 — checkpoint 恢复与渠道会话列表排序

- persist: formatTime 定宽;loadLatestSQL 补 rowid 兜底;New 时归一化历史行(避免 LoadLatest 挑到更旧检查点)
- channel/session: 写入定宽;listSessionSQL 补 rowid 兜底;核实无 INSERT 吃 DDL 默认值,故只加注释不改默认值
- 新增回归测试并做变异验证(改回 RFC3339Nano 即 FAIL)
This commit is contained in:
AskaEth committed 2026-10-03 16:08:51 +08:00
1 parent 31086ed1ab
commit fbbfcc95ee
7 files changed
+445 -14

No files matched your search

+11 -3
View File
@@ -19,6 +19,11 @@ import (
// channel maps to exactly one internal session_id. created_at/updated_at follow
// the project's timestamp convention; actor_role records the identity lane so the
// mapping survives restarts.
//
// The created_at/updated_at DEFAULT expressions are a fallback only: every write
// path (insertOrIgnoreSessionSQL) supplies explicit fixed-width UTC timestamps
// via formatSessionTime, so the 3-digit-millisecond strftime defaults are never
// exercised and no row mixes timestamp widths.
const createTableSQL = `
CREATE TABLE IF NOT EXISTS channel_sessions (
platform TEXT NOT NULL,
@@ -43,10 +48,13 @@ SELECT session_id FROM channel_sessions WHERE platform = ? AND channel_id = ?`
// listSessionsSQL selects every persisted channel mapping, newest first. Group
// sessions key on a channel_id carrying groupChannelPrefix; direct sessions use
// their channel id.
// their channel id. created_at is fixed-width UTC text (see sessionTimeLayout),
// so its text order is its time order; rowid breaks created_at ties so mappings
// written in the same coarse clock tick still list in a stable order
// (channel_sessions is an ordinary rowid table, not WITHOUT ROWID).
const listSessionSQL = `
SELECT platform, channel_id, session_id, actor_role FROM channel_sessions
ORDER BY created_at DESC`
ORDER BY created_at DESC, rowid DESC`
// groupChannelPrefix marks a channel id as a group conversation so a session
// reconstructed from the backing table (where no IsGroup flag is persisted) can
@@ -178,7 +186,7 @@ func (m *SessionManager) lookupCached(key string) (Session, bool) {
// winning writer's session_id becomes the binding. It is called under the write
// lock so the in-memory cache and the table stay in agreement.
func (m *SessionManager) persist(ctx context.Context, platform, cid string, actor adapter.Actor) (string, error) {
now := time.Now().UTC().Format(time.RFC3339Nano)
now := formatSessionTime(time.Now())
if _, err := m.db.ExecContext(ctx, insertOrIgnoreSessionSQL, platform, cid, newSessionID(), string(actor.Role), now, now); err != nil {
return "", fmt.Errorf("session: persist %s/%s: %w", platform, cid, err)
}
+95
View File
@@ -318,3 +318,98 @@ func TestNewRejectsNilDB(t *testing.T) {
t.Fatalf("New(nil db) error = %v, want ErrNilDB", err)
}
}
// insertSessionAt writes a channel_sessions row directly with a caller-supplied
// raw created_at/updated_at, standing in for a mapping written by past code.
func insertSessionAt(t *testing.T, handle *sql.DB, cid, at string) {
t.Helper()
if _, err := handle.ExecContext(context.Background(),
`INSERT INTO channel_sessions (platform, channel_id, session_id, actor_role, created_at, updated_at)
VALUES ('qq', ?, ?, 'owner', ?, ?)`,
cid, "sess-"+cid, at, at); err != nil {
t.Fatalf("insert %s: %v", cid, err)
}
}
// TestListNewestFirstDeterministic covers the ordering contract: three mappings
// written within the same coarse clock tick share one created_at, so only the
// rowid tiebreak can define their order. List is newest first, so the row
// inserted last (highest rowid) must come first. The cache is deliberately empty
// so List reads the persisted order instead of cached entries.
func TestListNewestFirstDeterministic(t *testing.T) {
t.Parallel()
m, handle := newTestManager(t)
ctx := context.Background()
const at = "2026-09-07T03:20:05.830000000Z"
for _, cid := range []string{"direct-a", "direct-b", "direct-c"} {
insertSessionAt(t, handle, cid, at)
}
got, err := m.List(ctx)
if err != nil {
t.Fatalf("List = %v", err)
}
want := []string{"direct-c", "direct-b", "direct-a"}
if len(got) != len(want) {
t.Fatalf("List = %d sessions, want %d", len(got), len(want))
}
for i, w := range want {
if got[i].ChannelID != w {
t.Errorf("got[%d].ChannelID = %q, want %q (rowid DESC tiebreak)", i, got[i].ChannelID, w)
}
}
}
// TestListNewestFirstByCreatedAt checks the created_at DESC half: the mapping
// with the latest fixed-width timestamp is listed first even though it was
// inserted first and therefore has the lowest rowid.
func TestListNewestFirstByCreatedAt(t *testing.T) {
t.Parallel()
m, handle := newTestManager(t)
ctx := context.Background()
insertSessionAt(t, handle, "direct-newest", "2026-09-07T03:20:05.830978900Z")
insertSessionAt(t, handle, "direct-oldest", "2026-09-07T03:20:05.830000000Z")
got, err := m.List(ctx)
if err != nil {
t.Fatalf("List = %v", err)
}
if len(got) != 2 {
t.Fatalf("List = %d sessions, want 2", len(got))
}
if got[0].ChannelID != "direct-newest" || got[1].ChannelID != "direct-oldest" {
t.Errorf("List order = [%q %q], want newest first", got[0].ChannelID, got[1].ChannelID)
}
}
// TestPersistWritesFixedWidthTimestamps pins the write path: GetOrCreate must
// store created_at/updated_at with the fixed-width layout so channel_sessions
// text ordering matches time ordering. RFC3339Nano would drop trailing zeros
// and persist a variable-width value instead.
func TestPersistWritesFixedWidthTimestamps(t *testing.T) {
t.Parallel()
m, handle := newTestManager(t)
ctx := context.Background()
if _, err := m.GetOrCreate(ctx, directMsg("o-100"), testOwner()); err != nil {
t.Fatalf("GetOrCreate = %v", err)
}
var created, updated string
if err := handle.QueryRowContext(ctx,
`SELECT created_at, updated_at FROM channel_sessions WHERE channel_id = ?`,
"direct-o-100").Scan(&created, &updated); err != nil {
t.Fatalf("read stored timestamps: %v", err)
}
if len(created) != sessionTimeLen || len(updated) != sessionTimeLen {
t.Errorf("stored timestamps = %q/%q (%d/%d bytes), want fixed width %d",
created, updated, len(created), len(updated), sessionTimeLen)
}
if created != updated {
t.Errorf("created_at %q != updated_at %q, want the same value on insert", created, updated)
}
}
+21
View File
@@ -0,0 +1,21 @@
package session
import "time"
// sessionTimeLayout is the fixed-width timestamp layout used for every stored
// created_at/updated_at value. Unlike time.RFC3339Nano it never drops trailing
// fractional zeros, so every value is exactly sessionTimeLen bytes wide. SQLite
// compares these TEXT columns as plain strings, and equal width is what makes
// that comparison agree with chronological order. Values are written in UTC, so
// the Z07:00 zone renders as a single "Z".
const sessionTimeLayout = "2006-01-02T15:04:05.000000000Z07:00"
// sessionTimeLen is the byte length of a UTC value formatted with
// sessionTimeLayout: "2006-01-02T15:04:05.000000000Z" — the Z07:00 zone
// collapses to "Z" in UTC.
const sessionTimeLen = 30
// formatSessionTime encodes t in UTC with the fixed-width sessionTimeLayout.
func formatSessionTime(t time.Time) string {
return t.UTC().Format(sessionTimeLayout)
}
+10 -11
View File
@@ -94,9 +94,13 @@ const (
loadSQL = `SELECT id, agent_id, step, state, created_at, updated_at
FROM checkpoints WHERE id = ?`
// updated_at is stored as fixed-width UTC text (see formatTime), so its text
// order is its time order. step breaks ties between checkpoints written in
// the same tick, and rowid makes that tiebreak deterministic (checkpoints is
// an ordinary rowid table, not WITHOUT ROWID).
loadLatestSQL = `SELECT id, agent_id, step, state, created_at, updated_at
FROM checkpoints WHERE agent_id = ?
ORDER BY updated_at DESC, step DESC LIMIT 1`
ORDER BY updated_at DESC, step DESC, rowid DESC LIMIT 1`
listSQL = `SELECT id, agent_id, step, state, created_at, updated_at
FROM checkpoints WHERE agent_id = ?
@@ -112,6 +116,8 @@ const (
// which already constrains the pool to a single connection for write
// serialization. The caller owns db and must Close it when done. log may be nil
// and is replaced with a discard logger so the Store can always log safely.
// Legacy variable-width timestamps are normalized in place (see
// normalizeTimestamps) so text ordering matches time ordering.
func New(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
if db == nil {
return nil, fmt.Errorf("persist: new: %w", ErrNilDB)
@@ -124,6 +130,9 @@ func New(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
if _, err := db.ExecContext(ctx, createIndexSQL); err != nil {
return nil, fmt.Errorf("persist: create index: %w", err)
}
if err := normalizeTimestamps(ctx, db); err != nil {
return nil, err
}
return &Store{db: db, log: log}, nil
}
@@ -270,16 +279,6 @@ func scanCheckpoint(sc rowScanner) (*Checkpoint, error) {
}, nil
}
// formatTime renders t as a UTC RFC3339 timestamp with nanosecond precision.
func formatTime(t time.Time) string {
return t.UTC().Format(time.RFC3339Nano)
}
// parseTime parses a timestamp written by formatTime.
func parseTime(s string) (time.Time, error) {
return time.Parse(time.RFC3339Nano, s)
}
// discardLogger returns log if non-nil, otherwise a logger that writes nowhere,
// so the Store can always log safely.
func discardLogger(log *slog.Logger) *slog.Logger {
+64
View File
@@ -360,3 +360,67 @@ func TestStore_Save_MissingID_异常(t *testing.T) {
t.Fatalf("Save(blank id) err = %v, want wrap of ErrMissingID", err)
}
}
// TestStore_LoadLatest_变宽时间戳_取较晚 is the regression for the variable-width
// timestamp bug on the durable-checkpoint recovery path. time.RFC3339Nano drops
// trailing fractional zeros, so the older checkpoint was once stored as
// ":05.83Z" and the later one as ":05.8309789Z". SQLite compares updated_at as
// plain text, and "…83Z" ranks above "…8309789Z", so LoadLatest used to recover
// the older checkpoint — a silent wrong resume. The older checkpoint also
// carries the larger step, so the step DESC tiebreak cannot mask the bad order.
func TestStore_LoadLatest_变宽时间戳_取较晚(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newTestStore(t)
const (
olderRaw = "2026-09-07T03:20:05.83Z" // X.8300000 under RFC3339Nano
newerRaw = "2026-09-07T03:20:05.8309789Z" // X.8309789 under RFC3339Nano
)
if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL,
"cp-older", "agent-a", 9, `{"step":9}`, olderRaw, olderRaw); err != nil {
t.Fatalf("insert older checkpoint: %v", err)
}
if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL,
"cp-newer", "agent-a", 2, `{"step":2}`, newerRaw, newerRaw); err != nil {
t.Fatalf("insert newer checkpoint: %v", err)
}
// Re-open so the fixed-width normalization repairs the legacy rows.
if _, err := New(ctx, s.db, testLogger()); err != nil {
t.Fatalf("New (normalize): %v", err)
}
got, err := s.LoadLatest(ctx, "agent-a")
if err != nil {
t.Fatalf("LoadLatest error: %v", err)
}
if got.ID != "cp-newer" {
t.Fatalf("LoadLatest = %q (step %d), want cp-newer (the later updated_at)", got.ID, got.Step)
}
if want := time.Date(2026, 9, 7, 3, 20, 5, 830978900, time.UTC); !got.UpdatedAt.Equal(want) {
t.Fatalf("LoadLatest updated_at = %v, want %v", got.UpdatedAt, want)
}
}
// TestStore_Save_定宽时间戳 pins that every write uses the fixed-width layout, so
// text ordering equals time ordering for rows created by the current code.
func TestStore_Save_定宽时间戳(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newTestStore(t)
if err := s.Save(ctx, &Checkpoint{ID: "cp-1", AgentID: "agent-a", Step: 1, State: []byte(`{}`)}); err != nil {
t.Fatalf("Save error: %v", err)
}
var created, updated string
if err := s.db.QueryRowContext(ctx,
`SELECT created_at, updated_at FROM checkpoints WHERE id = ?`, "cp-1").
Scan(&created, &updated); err != nil {
t.Fatalf("read stored timestamps: %v", err)
}
if len(created) != storedTimeLen || len(updated) != storedTimeLen {
t.Errorf("stored timestamps = %q/%q (%d/%d bytes), want fixed width %d",
created, updated, len(created), len(updated), storedTimeLen)
}
}
+126
View File
@@ -0,0 +1,126 @@
package persist
import (
"context"
"database/sql"
"fmt"
"time"
)
// timeLayout is the fixed-width timestamp layout used for every stored
// created_at/updated_at value. Unlike time.RFC3339Nano it never drops trailing
// fractional zeros, so every value is exactly storedTimeLen bytes wide. SQLite
// compares these TEXT columns as plain strings, and equal width is what makes
// that comparison agree with chronological order. UTC values render the zone as
// a single "Z".
const timeLayout = "2006-01-02T15:04:05.000000000Z07:00"
// storedTimeLen is the byte length of a UTC value formatted with timeLayout:
// "2006-01-02T15:04:05.000000000Z" — the Z07:00 zone collapses to "Z" in UTC.
const storedTimeLen = 30
// checkpointsTable names the table whose timestamps are normalized when the
// Store opens. It must match the DDL in persist.go.
const checkpointsTable = "checkpoints"
// formatTime encodes t in UTC with the fixed-width timeLayout. Always writing
// UTC keeps ordering and round-trips stable across processes, and the fixed
// width is what lets ORDER BY updated_at order these TEXT columns by time.
func formatTime(t time.Time) string {
return t.UTC().Format(timeLayout)
}
// parseTime decodes a stored timestamp. time.RFC3339Nano accepts both the
// fixed-width values this package writes and the legacy variable-width ones, so
// the same parser serves reads and the open-time normalization.
func parseTime(s string) (time.Time, error) {
return time.Parse(time.RFC3339Nano, s)
}
// normalizeTimestamps rewrites every checkpoints timestamp that is not
// storedTimeLen bytes wide into the fixed-width timeLayout. Rows written before
// the layout existed carry variable-width RFC3339Nano text whose string order is
// not its time order; this brings them in line so LoadLatest picks the truly
// latest checkpoint.
//
// It is idempotent: only rows whose column length differs from storedTimeLen are
// selected, so running it again on a normalized table is a no-op. A row whose
// value does not parse is left exactly as it is — one bad legacy value must not
// keep the Store from opening.
func normalizeTimestamps(ctx context.Context, db *sql.DB) error {
if err := normalizeTimeColumn(ctx, db, "created_at"); err != nil {
return err
}
return normalizeTimeColumn(ctx, db, "updated_at")
}
// normalizeTimeColumn rewrites every checkpoints.<col> value that is not
// storedTimeLen bytes wide. col is a compile-time constant, never user input, so
// splicing it into the statement is safe; the length comparison and every value
// written are parameterized. All rewrites are collected before any UPDATE runs,
// so no write happens while the SELECT cursor is still open.
func normalizeTimeColumn(ctx context.Context, db *sql.DB, col string) error {
fixes, err := collectLegacyTimestamps(ctx, db, col)
if err != nil {
return err
}
for _, f := range fixes {
if _, err := db.ExecContext(ctx,
"UPDATE "+checkpointsTable+" SET "+f.col+" = ? WHERE id = ?", f.value, f.id); err != nil {
return fmt.Errorf("persist: normalize %s.%s: %w", checkpointsTable, f.col, err)
}
}
return nil
}
// timeFix is one pending timestamp rewrite: which column, which row (by primary
// key id), and the fixed-width value to write.
type timeFix struct {
col string
id string
value string
}
// collectLegacyTimestamps reads the rows of checkpoints.col whose value is not
// storedTimeLen bytes wide and returns the parseable ones as pending rewrites.
// Unparseable values are skipped, so normalization can never fail a row: the only
// errors are genuine read failures.
func collectLegacyTimestamps(ctx context.Context, db *sql.DB, col string) ([]timeFix, error) {
rows, err := db.QueryContext(ctx,
"SELECT id, "+col+" FROM "+checkpointsTable+" WHERE length("+col+") != ?", storedTimeLen)
if err != nil {
return nil, fmt.Errorf("persist: query legacy %s.%s: %w", checkpointsTable, col, err)
}
defer func() { _ = rows.Close() }()
fixes := make([]timeFix, 0)
for rows.Next() {
var id, raw string
if err := rows.Scan(&id, &raw); err != nil {
return nil, fmt.Errorf("persist: scan legacy %s.%s: %w", checkpointsTable, col, err)
}
fixed, ok := reformatTime(raw)
if !ok {
continue
}
fixes = append(fixes, timeFix{col: col, id: id, value: fixed})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("persist: iterate legacy %s.%s: %w", checkpointsTable, col, err)
}
return fixes, nil
}
// reformatTime re-encodes raw with timeLayout. The bool is false when raw is
// already storedTimeLen wide or when it does not parse, in which case the caller
// leaves the stored value untouched.
func reformatTime(raw string) (string, bool) {
if len(raw) == storedTimeLen {
return "", false
}
t, err := parseTime(raw)
if err != nil || t.IsZero() {
return "", false
}
return formatTime(t), true
}
+118
View File
@@ -0,0 +1,118 @@
package persist
import (
"context"
"testing"
"time"
)
// insertLegacyCheckpointSQL writes a checkpoints row with caller-supplied raw
// timestamps, standing in for a row persisted before the fixed-width layout.
const insertLegacyCheckpointSQL = `INSERT INTO checkpoints
(id, agent_id, step, state, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)`
func TestFormatTime_定宽_正常(t *testing.T) {
t.Parallel()
tests := []struct {
name string
at time.Time
want string
}{
{"零小数秒", time.Date(2026, 9, 7, 3, 20, 5, 0, time.UTC), "2026-09-07T03:20:05.000000000Z"},
{"毫秒", time.Date(2026, 9, 7, 3, 20, 5, 830000000, time.UTC), "2026-09-07T03:20:05.830000000Z"},
{"纳秒", time.Date(2026, 9, 7, 3, 20, 5, 830978900, time.UTC), "2026-09-07T03:20:05.830978900Z"},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
got := formatTime(tt.at)
if got != tt.want {
t.Errorf("formatTime = %q, want %q", got, tt.want)
}
if len(got) != storedTimeLen {
t.Errorf("formatTime length = %d, want %d", len(got), storedTimeLen)
}
})
}
}
func TestFormatTime_非UTC_正常(t *testing.T) {
t.Parallel()
zone := time.FixedZone("UTC+8", 8*60*60)
// 11:20:05.830+08:00 is 03:20:05.830Z, so the stored text is UTC.
at := time.Date(2026, 9, 7, 11, 20, 5, 830000000, zone)
if got, want := formatTime(at), "2026-09-07T03:20:05.830000000Z"; got != want {
t.Errorf("formatTime = %q, want %q", got, want)
}
}
// TestNew_归一化历史时间戳_正常 writes a legacy variable-width row directly,
// re-opens the Store to trigger normalization, and checks both the repaired text
// and its idempotence on a further open.
func TestNew_归一化历史时间戳_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newTestStore(t)
// RFC3339Nano drops the trailing fractional zeros, so this legacy value is
// only 23 bytes wide and loses the text-ordering guarantee.
const legacy = "2026-09-07T03:20:05.83Z"
if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL,
"legacy", "agent-a", 1, `{}`, legacy, legacy); err != nil {
t.Fatalf("insert legacy row: %v", err)
}
if _, err := New(ctx, s.db, testLogger()); err != nil {
t.Fatalf("New (normalize): %v", err)
}
var created, updated string
if err := s.db.QueryRowContext(ctx,
`SELECT created_at, updated_at FROM checkpoints WHERE id = ?`, "legacy").
Scan(&created, &updated); err != nil {
t.Fatalf("read normalized timestamps: %v", err)
}
const want = "2026-09-07T03:20:05.830000000Z"
if created != want || updated != want {
t.Errorf("normalized timestamps = %q/%q, want %q", created, updated, want)
}
// A further open must be a no-op: the length condition selects nothing once
// the row is fixed width.
if _, err := New(ctx, s.db, testLogger()); err != nil {
t.Fatalf("New (second pass): %v", err)
}
var again string
if err := s.db.QueryRowContext(ctx,
`SELECT created_at FROM checkpoints WHERE id = ?`, "legacy").Scan(&again); err != nil {
t.Fatalf("read created_at after second open: %v", err)
}
if again != created {
t.Errorf("second open changed created_at: %q -> %q", created, again)
}
}
// TestNew_无法解析的时间戳_跳过 pins the "leave it alone" rule: one bad legacy
// value must not keep the Store from opening, and must not be rewritten.
func TestNew_无法解析的时间戳_跳过(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newTestStore(t)
if _, err := s.db.ExecContext(ctx, insertLegacyCheckpointSQL,
"broken", "agent-a", 1, `{}`, "not-a-time", "not-a-time"); err != nil {
t.Fatalf("insert unparseable row: %v", err)
}
if _, err := New(ctx, s.db, testLogger()); err != nil {
t.Fatalf("New with unparseable timestamp: %v", err)
}
var raw string
if err := s.db.QueryRowContext(ctx,
`SELECT created_at FROM checkpoints WHERE id = ?`, "broken").Scan(&raw); err != nil {
t.Fatalf("read created_at: %v", err)
}
if raw != "not-a-time" {
t.Errorf("created_at = %q, want the unparseable value left as is", raw)
}
}