From a4e51ec5cd688e451d1e7734b2d3c7036ac46b51 Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 15:47:32 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8C=89=E5=AE=8C=E6=88=90=E6=97=B6?= =?UTF-8?q?=E5=88=BB=E5=88=86=E6=89=B9=E6=B8=85=E7=90=86=E5=B9=B6=E9=99=90?= =?UTF-8?q?=E5=88=B6=E8=AF=BB=E8=BF=9E=E6=8E=A5=E6=B1=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/app/message/ack.go | 4 +- internal/app/message/dispatch.go | 4 +- internal/app/message/recover.go | 9 +- internal/app/message/review_c03_test.go | 271 ++++++++++++++++++ internal/store/db.go | 6 + internal/store/db_pool_test.go | 48 ++++ internal/store/db_test.go | 12 +- internal/store/migrate_test.go | 2 +- .../store/migrations/0004_cleanup_indexes.sql | 11 + 9 files changed, 350 insertions(+), 17 deletions(-) create mode 100644 internal/app/message/review_c03_test.go create mode 100644 internal/store/db_pool_test.go create mode 100644 internal/store/migrations/0004_cleanup_indexes.sql diff --git a/internal/app/message/ack.go b/internal/app/message/ack.go index 0be436b..220ee3e 100644 --- a/internal/app/message/ack.go +++ b/internal/app/message/ack.go @@ -107,8 +107,8 @@ func (a *App) Recall(ctx context.Context, senderID string, req *protocol.Recall) return err } if state == StateScheduled { - if _, e := tx.Exec(`UPDATE messages SET state = ?, reason = ? WHERE seq = ? AND state = 'scheduled'`, - StateCompleted, ReasonRecalled, seq); e != nil { + if _, e := tx.Exec(`UPDATE messages SET state = ?, reason = ?, completed_at = ? WHERE seq = ? AND state = 'scheduled'`, + StateCompleted, ReasonRecalled, nowMs, seq); e != nil { return e } if _, e := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); e != nil { diff --git a/internal/app/message/dispatch.go b/internal/app/message/dispatch.go index 4ad1775..e96489a 100644 --- a/internal/app/message/dispatch.go +++ b/internal/app/message/dispatch.go @@ -239,8 +239,8 @@ func FinalizeMessageTx(tx *sql.Tx, seq int64, wantReceipt bool, senderID, endpoi return err } } - if _, err := tx.Exec(`UPDATE messages SET state = ?, reason = CASE WHEN ? != '' THEN ? ELSE reason END WHERE seq = ?`, - StateCompleted, msgReason, msgReason, seq); err != nil { + if _, err := tx.Exec(`UPDATE messages SET state = ?, reason = CASE WHEN ? != '' THEN ? ELSE reason END, completed_at = ? WHERE seq = ?`, + StateCompleted, msgReason, msgReason, nowMs, seq); err != nil { return err } if _, err := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); err != nil { diff --git a/internal/app/message/recover.go b/internal/app/message/recover.go index e01e6ac..e65b3db 100644 --- a/internal/app/message/recover.go +++ b/internal/app/message/recover.go @@ -188,15 +188,12 @@ func (a *App) purgeCompletedMessages(ctx context.Context, nowMs int64) error { var seqs []int64 err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error { q := ` -SELECT m.seq FROM messages m -WHERE m.state = 'completed'` +SELECT seq FROM messages +WHERE state = 'completed'` args := []any{} if days > 0 { cutoff := nowMs - int64(days)*24*3600*1000 - q += ` AND COALESCE( - (SELECT MAX(d.updated_at) FROM deliveries d WHERE d.seq = m.seq), - m.send_at -) < ?` + q += ` AND COALESCE(completed_at, send_at) < ?` args = append(args, cutoff) } q += ` LIMIT ?` diff --git a/internal/app/message/review_c03_test.go b/internal/app/message/review_c03_test.go new file mode 100644 index 0000000..fe32383 --- /dev/null +++ b/internal/app/message/review_c03_test.go @@ -0,0 +1,271 @@ +package message + +import ( + "context" + "database/sql" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.asio.asia/nixevol/NixMsg/internal/app/port" + "git.asio.asia/nixevol/NixMsg/internal/store" +) + +func TestC03PurgePlansUseIndexes(t *testing.T) { + t.Parallel() + e := openDeliveryEnv(t, nil) + ctx := context.Background() + var plan string + rows, err := e.db.Read.QueryContext(ctx, `EXPLAIN QUERY PLAN DELETE FROM receipts WHERE receipt_id IN ( + SELECT receipt_id FROM receipts WHERE created_at < 0 LIMIT 2000)`) + if err != nil { + t.Fatal(err) + } + plan = collectPlan(t, rows) + if !strings.Contains(plan, "idx_receipts_created") { + t.Fatalf("receipts plan missing idx_receipts_created: %s", plan) + } + if strings.Contains(plan, "SCAN receipts") && !strings.Contains(plan, "USING INDEX") { + t.Fatalf("SCAN receipts without index: %s", plan) + } + + rows, err = e.db.Read.QueryContext(ctx, `EXPLAIN QUERY PLAN DELETE FROM send_keys WHERE rowid IN ( + SELECT sk.rowid FROM send_keys sk + WHERE sk.created_at < 0 + AND NOT EXISTS (SELECT 1 FROM messages m WHERE m.sender_id = sk.sender_id AND m.id = sk.msg_id) + LIMIT 2000)`) + if err != nil { + t.Fatal(err) + } + plan = collectPlan(t, rows) + if strings.Contains(plan, "SCAN sk") && !strings.Contains(plan, "USING INDEX") { + t.Fatalf("SCAN sk without index: %s", plan) + } + + rows, err = e.db.Read.QueryContext(ctx, `EXPLAIN QUERY PLAN +UPDATE deliveries SET pushed_conn = NULL WHERE endpoint_id = 'x' AND state = 'pending' AND pushed_conn = 'y'`) + if err != nil { + t.Fatal(err) + } + plan = collectPlan(t, rows) + if !strings.Contains(plan, "idx_deliveries_outbox") { + t.Fatalf("disconnect plan: %s", plan) + } + + rows, err = e.db.Read.QueryContext(ctx, `EXPLAIN QUERY PLAN SELECT seq FROM messages WHERE completed_at < 0`) + if err != nil { + t.Fatal(err) + } + plan = collectPlan(t, rows) + if !strings.Contains(plan, "idx_messages_completed_at") { + t.Fatalf("completed_at plan missing idx_messages_completed_at: %s", plan) + } +} + +func collectPlan(t *testing.T, rows *sql.Rows) string { + t.Helper() + defer func() { _ = rows.Close() }() + var b strings.Builder + cols, _ := rows.Columns() + vals := make([]any, len(cols)) + ptrs := make([]any, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + for rows.Next() { + if err := rows.Scan(ptrs...); err != nil { + t.Fatal(err) + } + for i, v := range vals { + if i > 0 { + b.WriteByte(' ') + } + switch x := v.(type) { + case []byte: + b.Write(x) + case string: + b.WriteString(x) + } + } + b.WriteByte('\n') + } + return b.String() +} + +func TestC03ExpireDoesNotPurgeReceipts(t *testing.T) { + t.Parallel() + e := openDeliveryEnv(t, func(l *Limits) { l.ReceiptRetentionDays = 1 }) + insertEndpoint(t, e.db, "alice", "", 1, 0) + ctx := context.Background() + old := e.nowMs - 2*24*3600*1000 + err := e.db.Queue.Do(ctx, func(tx *sql.Tx) error { + _, err := tx.Exec(`INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked) +VALUES('alice','m1','bob','accepted','',?,0)`, old) + return err + }) + if err != nil { + t.Fatal(err) + } + if err := e.app.ExpireOnce(ctx, e.nowMs); err != nil { + t.Fatal(err) + } + var n int + _ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM receipts`).Scan(&n) + if n != 1 { + t.Fatalf("expire deleted receipts: %d", n) + } + if err := e.app.PurgeOnce(ctx, e.nowMs); err != nil { + t.Fatal(err) + } + _ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM receipts`).Scan(&n) + if n != 0 { + t.Fatalf("purge left receipts: %d", n) + } +} + +func TestC03RetentionZeroDeletesCompleted(t *testing.T) { + t.Parallel() + e := openDeliveryEnv(t, func(l *Limits) { l.RecordRetentionDays = 0 }) + insertEndpoint(t, e.db, "alice", "", 1, 0) + ctx := context.Background() + err := e.db.Queue.Do(ctx, func(tx *sql.Tx) error { + _, err := tx.Exec(`INSERT INTO messages(id, sender_id, dest_kind, dest_id, meta, content_type, body_enc, send_at, keep, ttl_seconds, receipt, state, reason, created_at) +VALUES('c0','alice','endpoint','bob','{}','text/plain','utf8',?,?,0,0,'completed','',?)`, e.nowMs, e.nowMs, e.nowMs) + return err + }) + if err != nil { + t.Fatal(err) + } + if err := e.app.PurgeOnce(ctx, e.nowMs); err != nil { + t.Fatal(err) + } + var n int + _ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE state='completed'`).Scan(&n) + if n != 0 { + t.Fatalf("completed left=%d", n) + } +} + +func TestC03ZombieExpireThenDrop(t *testing.T) { + t.Parallel() + e := openDeliveryEnv(t, nil) + insertEndpoint(t, e.db, "alice", "", 1, 0) + insertEndpoint(t, e.db, "bob", "", 1, 0) + _ = e.db.Queue.Do(context.Background(), func(tx *sql.Tx) error { + _, err := tx.Exec(`UPDATE endpoints SET online_since=?, offline_since=NULL WHERE id=?`, e.nowMs, "bob") + return err + }) + e.conns.Set("bob", LiveConn{ConnID: "stale"}) // 未就绪：分发仍算在线，但清理视为无就绪连接 + ctx := context.Background() + if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("zom1", "bob")); err != nil { + t.Fatal(err) + } + seq := e.seqOf("alice", "zom1") + var exp sql.NullInt64 + _ = e.db.Read.QueryRow(`SELECT expire_at FROM deliveries WHERE seq=?`, seq).Scan(&exp) + if exp.Valid { + t.Fatalf("online handshake-in-progress should leave expire_at empty, got %v", exp) + } + _ = e.app.OnDisconnect(ctx, "bob", "stale", false) + if err := e.app.ExpireOnce(ctx, e.nowMs); err != nil { + t.Fatal(err) + } + _ = e.db.Read.QueryRow(`SELECT expire_at FROM deliveries WHERE seq=?`, seq).Scan(&exp) + if !exp.Valid { + t.Fatal("zombie expire_at not filled") + } + e.setNow(exp.Int64 + 1) + if err := e.app.ExpireOnce(ctx, e.nowMs); err != nil { + t.Fatal(err) + } + st, reason := e.deliveryState(seq, "bob") + if st != DeliveryDropped || reason != ReasonOffline { + t.Fatalf("got %s/%s", st, reason) + } +} + +func TestC03WALShrinksAfterPurge(t *testing.T) { + t.Parallel() + dir := t.TempDir() + data := filepath.Join(dir, "data") + db, err := store.Open(data, "FULL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + ctx := context.Background() + nowMs := int64(1_700_000_000_000) + err = db.Queue.Do(ctx, func(tx *sql.Tx) error { + if _, err := tx.Exec(`INSERT INTO endpoints(id,name,login_hash,enabled,created_at) VALUES('alice','a','x',1,?)`, nowMs); err != nil { + return err + } + for i := 0; i < 200; i++ { + if _, err := tx.Exec(`INSERT INTO messages(id, sender_id, dest_kind, dest_id, meta, content_type, body_enc, send_at, keep, ttl_seconds, receipt, state, reason, created_at) +VALUES(?,?, 'endpoint','bob','{}','text/plain','utf8',?,0,0,0,'completed','',?)`, + "m"+itoa(i), "alice", nowMs-10*24*3600*1000, nowMs-10*24*3600*1000); err != nil { + return err + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + walPath := filepath.Join(data, store.DBFileName+"-wal") + before, _ := os.Stat(walPath) + app := New(db, Limits{RecordRetentionDays: 1, ReceiptRetentionDays: 1, IdempotencyHours: 1}, nil) + app.nowFn = func() time.Time { return time.UnixMilli(nowMs) } + if err := app.PurgeOnce(ctx, nowMs); err != nil { + t.Fatal(err) + } + after, err := os.Stat(walPath) + if err == nil && before != nil && after.Size() > before.Size() { + t.Fatalf("wal grew %d -> %d", before.Size(), after.Size()) + } + var n int + _ = db.Read.QueryRow(`SELECT COUNT(*) FROM messages`).Scan(&n) + if n != 0 { + t.Fatalf("messages left=%d", n) + } +} + +func TestC03DispatchOfflineStillOnlineInDB(t *testing.T) { + t.Parallel() + e := openDeliveryEnv(t, nil) + insertEndpoint(t, e.db, "alice", "", 1, 0) + insertEndpoint(t, e.db, "bob", "", 1, 0) + _ = e.db.Queue.Do(context.Background(), func(tx *sql.Tx) error { + _, err := tx.Exec(`UPDATE endpoints SET online_since=?, offline_since=NULL WHERE id=?`, e.nowMs-1000, "bob") + return err + }) + ctx := context.Background() + if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("dbon", "bob")); err != nil { + t.Fatal(err) + } + seq := e.seqOf("alice", "dbon") + st, _ := e.deliveryState(seq, "bob") + if st != DeliveryPending { + t.Fatalf("state=%s", st) + } + var exp sql.NullInt64 + _ = e.db.Read.QueryRow(`SELECT expire_at FROM deliveries WHERE seq=?`, seq).Scan(&exp) + if !exp.Valid || exp.Int64 != e.nowMs+60_000 { + t.Fatalf("expire_at=%v want now+grace", exp) + } +} + +func TestC03ReadPoolAndQueryOnly(t *testing.T) { + t.Parallel() + dir := t.TempDir() + db, err := store.Open(filepath.Join(dir, "data"), "FULL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + ctx := context.Background() + if _, err := db.Read.ExecContext(ctx, `INSERT INTO settings(key,value,updated_at) VALUES('k','v',1)`); err == nil { + t.Fatal("read connection accepted write") + } +} diff --git a/internal/store/db.go b/internal/store/db.go index 12faf93..7a6577b 100644 --- a/internal/store/db.go +++ b/internal/store/db.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "time" _ "modernc.org/sqlite" ) @@ -134,6 +135,9 @@ func OpenReader(dataDir, synchronous string) (*sql.DB, error) { if err != nil { return nil, fmt.Errorf("open sqlite reader: %w", err) } + db.SetMaxOpenConns(64) + db.SetMaxIdleConns(16) + db.SetConnMaxIdleTime(5 * time.Minute) if err := db.Ping(); err != nil { _ = db.Close() return nil, fmt.Errorf("ping sqlite reader: %w", err) @@ -157,6 +161,8 @@ func buildDSN(dataDir, synchronous string, writer bool) (string, error) { ) if writer { dsn += "&_txlock=immediate" + } else { + dsn += "&_pragma=query_only(1)" } return dsn, nil } diff --git a/internal/store/db_pool_test.go b/internal/store/db_pool_test.go new file mode 100644 index 0000000..cfff3a7 --- /dev/null +++ b/internal/store/db_pool_test.go @@ -0,0 +1,48 @@ +package store + +import ( + "context" + "sync" + "sync/atomic" + "testing" +) + +func TestReaderPoolCapAndQueryOnly(t *testing.T) { + t.Parallel() + dir := t.TempDir() + db, err := Open(dir, "FULL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + + ctx := context.Background() + if _, err := db.Read.ExecContext(ctx, `INSERT INTO settings(key,value,updated_at) VALUES('x','y',1)`); err == nil { + t.Fatal("query_only reader accepted write") + } + + var maxOpen atomic.Int32 + var wg sync.WaitGroup + for i := 0; i < 200; i++ { + wg.Add(1) + go func() { + defer wg.Done() + var one int + if qErr := db.Read.QueryRowContext(ctx, `SELECT 1`).Scan(&one); qErr != nil { + t.Errorf("read: %v", qErr) + return + } + n := int32(db.Read.Stats().OpenConnections) + for { + cur := maxOpen.Load() + if n <= cur || maxOpen.CompareAndSwap(cur, n) { + break + } + } + }() + } + wg.Wait() + if maxOpen.Load() > 64 { + t.Fatalf("open connections %d exceed 64", maxOpen.Load()) + } +} diff --git a/internal/store/db_test.go b/internal/store/db_test.go index 6b5a464..8864f20 100644 --- a/internal/store/db_test.go +++ b/internal/store/db_test.go @@ -19,7 +19,7 @@ func TestOpenEmptyDirCreatesTables(t *testing.T) { } defer func() { _ = db.Close() }() - assertMigrationCount(t, db.Write, 2) + assertMigrationCount(t, db.Write, 3) for _, table := range []string{ "endpoints", "settings", "talk_grants", "groups", "group_members", "messages", "message_bodies", "deliveries", "receipts", "send_keys", @@ -41,7 +41,7 @@ func TestOpenIdempotentNoRemigrate(t *testing.T) { if err != nil { t.Fatal(err) } - assertMigrationCount(t, db1.Write, 2) + assertMigrationCount(t, db1.Write, 3) _ = db1.Close() db2, err := Open(dir, "FULL") @@ -49,7 +49,7 @@ func TestOpenIdempotentNoRemigrate(t *testing.T) { t.Fatal(err) } defer func() { _ = db2.Close() }() - assertMigrationCount(t, db2.Write, 2) + assertMigrationCount(t, db2.Write, 3) if entries, readErr := os.ReadDir(filepath.Join(dir, "backup")); readErr == nil && len(entries) > 0 { t.Fatalf("idempotent reopen should not backup: %v", entries) @@ -83,7 +83,7 @@ CREATE TABLE schema_migrations ( } defer func() { _ = db.Close() }() - assertMigrationCount(t, db.Write, 2) + assertMigrationCount(t, db.Write, 3) assertTableExists(t, db.Write, "endpoints") assertTableExists(t, db.Write, "api_tokens") @@ -96,7 +96,7 @@ CREATE TABLE schema_migrations ( var names []string for _, e := range entries { names = append(names, e.Name()) - if e.Name() == "pre-migrate-v1-to-v2.db" { + if e.Name() == "pre-migrate-v1-to-v4.db" { found = true info, statErr := e.Info() if statErr != nil { @@ -108,7 +108,7 @@ CREATE TABLE schema_migrations ( } } if !found { - t.Fatalf("expected pre-migrate-v1-to-v2.db, got %v", names) + t.Fatalf("expected pre-migrate-v1-to-v4.db, got %v", names) } } diff --git a/internal/store/migrate_test.go b/internal/store/migrate_test.go index 4e47f15..f26d6b6 100644 --- a/internal/store/migrate_test.go +++ b/internal/store/migrate_test.go @@ -36,7 +36,7 @@ func TestFailingMigrationReusesBackup(t *testing.T) { if err != nil { t.Fatal(err) } - if len(first) != 1 || first[0].Name() != "pre-migrate-v2-to-v9999.db" { + if len(first) != 1 || first[0].Name() != "pre-migrate-v4-to-v9999.db" { t.Fatalf("first backups=%v", dirNames(first)) } diff --git a/internal/store/migrations/0004_cleanup_indexes.sql b/internal/store/migrations/0004_cleanup_indexes.sql new file mode 100644 index 0000000..3dff3f0 --- /dev/null +++ b/internal/store/migrations/0004_cleanup_indexes.sql @@ -0,0 +1,11 @@ +-- C-03：完成时刻列与清理索引（编号避开 U-02 的 0003） +ALTER TABLE messages ADD COLUMN completed_at INTEGER; + +UPDATE messages SET completed_at = COALESCE( + (SELECT MAX(d.updated_at) FROM deliveries d WHERE d.seq = messages.seq), + send_at +) WHERE state = 'completed' AND completed_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_messages_completed_at ON messages(completed_at); +CREATE INDEX IF NOT EXISTS idx_receipts_created ON receipts(created_at); +CREATE INDEX IF NOT EXISTS idx_send_keys_created ON send_keys(created_at);