fix: 按完成时刻分批清理并限制读连接池

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'`,
+			if _, e := tx.Exec(`UPDATE messages SET state = ?, reason = ?, completed_at = ? WHERE seq = ? AND state = 'scheduled'`,
-				StateCompleted, ReasonRecalled, seq); e != nil {
+				StateCompleted, ReasonRecalled, nowMs, seq); e != nil {
 				return e
 			}
 			if _, e := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); e != nil {

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 = ?`,
+	if _, err := tx.Exec(`UPDATE messages SET state = ?, reason = CASE WHEN ? != '' THEN ? ELSE reason END, completed_at = ? WHERE seq = ?`,
-		StateCompleted, msgReason, msgReason, seq); err != nil {
+		StateCompleted, msgReason, msgReason, nowMs, seq); err != nil {
 		return err
 	}
 	if _, err := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); err != nil {

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
+SELECT seq FROM messages
-WHERE m.state = 'completed'`
+WHERE state = 'completed'`
 			args := []any{}
 			if days > 0 {
 				cutoff := nowMs - int64(days)*24*3600*1000
-				q += ` AND COALESCE(
+				q += ` AND COALESCE(completed_at, send_at) < ?`
-  (SELECT MAX(d.updated_at) FROM deliveries d WHERE d.seq = m.seq),
-  m.send_at
-) < ?`
 				args = append(args, cutoff)
 			}
 			q += ` LIMIT ?`

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")
+	}
+}

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
 }

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())
+	}
+}

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)
 	}
 }
 

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))
 	}
 

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);