修复日志持久化可靠性与数据库安全默认值

.env.example

@@ -56,7 +56,7 @@ CLICKHOUSE_USER=tokenthief
 CLICKHOUSE_PASSWORD=
 CLICKHOUSE_DB=tokenthief
 # ClickHouse 的宿主机监听 IP，以及 HTTP 和原生客户端端口
-CLICKHOUSE_LISTEN_IP=0.0.0.0
+CLICKHOUSE_LISTEN_IP=127.0.0.1
 CLICKHOUSE_HTTP_PORT=8123
 CLICKHOUSE_NATIVE_PORT=9000
 

compose.yml

@@ -43,8 +43,8 @@ services:
     container_name: tokenthief-clickhouse
     restart: unless-stopped
     ports:
-      - "${CLICKHOUSE_LISTEN_IP:-0.0.0.0}:${CLICKHOUSE_HTTP_PORT:-8123}:8123"
-      - "${CLICKHOUSE_LISTEN_IP:-0.0.0.0}:${CLICKHOUSE_NATIVE_PORT:-9000}:9000"
+      - "${CLICKHOUSE_LISTEN_IP:-127.0.0.1}:${CLICKHOUSE_HTTP_PORT:-8123}:8123"
+      - "${CLICKHOUSE_LISTEN_IP:-127.0.0.1}:${CLICKHOUSE_NATIVE_PORT:-9000}:9000"
     environment:
       CLICKHOUSE_USER: "${CLICKHOUSE_USER:-tokenthief}"
       CLICKHOUSE_PASSWORD: "${CLICKHOUSE_PASSWORD:?set a strong CLICKHOUSE_PASSWORD}"

db/clickhouse.go

@@ -25,6 +25,16 @@ type Pool struct {
 	generation        uint64
 	healthy           bool
 	closed            bool
+	leases            map[uint64]*connectionLease
+	leaseChanged      chan struct{}
+	closing           int
+}
+
+type connectionLease struct {
+	conn    clickhouse.Conn
+	active  int
+	retired bool
+	closed  bool
 }
 
 var errPoolClosed = errors.New("db pool is closed")
@@ -79,13 +89,16 @@ func (p *Pool) connect(ctx context.Context) error {
 		_ = conn.Close()
 		return errPoolClosed
 	}
-	if p.conn != nil {
-		_ = p.conn.Close()
-	}
+	old := p.retireCurrentLocked()
 	p.conn = conn
 	p.generation++
+	p.ensureLeaseLocked(p.generation, conn)
 	p.healthy = true
 	p.mu.Unlock()
+	if old != nil {
+		_ = old.Close()
+		p.finishClosing()
+	}
 	log.Printf("[db] connected and migrated")
 	return nil
 }
@@ -194,13 +207,14 @@ func (p *Pool) watch(ctx context.Context) {
 			return
 		case <-t.C:
 			if p.Healthy() {
-				if conn, generation := p.GetWithGeneration(); conn != nil {
+				if conn, generation, release := p.Acquire(); conn != nil {
 					pctx, cancel := context.WithTimeout(ctx, 3*time.Second)
 					if err := conn.Ping(pctx); err != nil {
 						log.Printf("[db] ping failed, marking unhealthy: %v", err)
 						p.MarkUnhealthyGeneration(generation)
 					}
 					cancel()
+					release()
 				}
 				continue
 			}
@@ -253,24 +267,126 @@ func (p *Pool) GetWithGeneration() (clickhouse.Conn, uint64) {
 	return p.conn, p.generation
 }
 
+// Acquire pins the current connection until release is called.
+func (p *Pool) Acquire() (clickhouse.Conn, uint64, func()) {
+	p.mu.Lock()
+	if p.closed || p.conn == nil {
+		p.mu.Unlock()
+		return nil, p.generation, nil
+	}
+	generation := p.generation
+	lease := p.ensureLeaseLocked(generation, p.conn)
+	lease.active++
+	p.mu.Unlock()
+
+	var once sync.Once
+	return lease.conn, generation, func() {
+		once.Do(func() { p.release(generation) })
+	}
+}
+
+func (p *Pool) ensureLeaseLocked(generation uint64, conn clickhouse.Conn) *connectionLease {
+	if p.leases == nil {
+		p.leases = make(map[uint64]*connectionLease)
+	}
+	lease := p.leases[generation]
+	if lease == nil {
+		lease = &connectionLease{conn: conn}
+		p.leases[generation] = lease
+	}
+	return lease
+}
+
+func (p *Pool) notifyLeaseChangedLocked() {
+	if p.leaseChanged != nil {
+		close(p.leaseChanged)
+	}
+	p.leaseChanged = make(chan struct{})
+}
+
+func (p *Pool) retireCurrentLocked() clickhouse.Conn {
+	if p.conn == nil {
+		return nil
+	}
+	lease := p.ensureLeaseLocked(p.generation, p.conn)
+	lease.retired = true
+	if lease.active != 0 || lease.closed {
+		return nil
+	}
+	lease.closed = true
+	delete(p.leases, p.generation)
+	p.closing++
+	return lease.conn
+}
+
+func (p *Pool) finishClosing() {
+	p.mu.Lock()
+	p.closing--
+	p.notifyLeaseChangedLocked()
+	p.mu.Unlock()
+}
+
+func (p *Pool) release(generation uint64) {
+	p.mu.Lock()
+	lease := p.leases[generation]
+	if lease == nil {
+		p.mu.Unlock()
+		return
+	}
+	lease.active--
+	var conn clickhouse.Conn
+	if lease.active == 0 && lease.retired && !lease.closed {
+		lease.closed = true
+		conn = lease.conn
+		p.closing++
+	}
+	p.mu.Unlock()
+	if conn != nil {
+		_ = conn.Close()
+		p.mu.Lock()
+		delete(p.leases, generation)
+		p.closing--
+		p.notifyLeaseChangedLocked()
+		p.mu.Unlock()
+	}
+}
+
 // Close starts closing the underlying connection and waits within ctx.
 func (p *Pool) Close(ctx context.Context) error {
 	p.mu.Lock()
 	p.closed = true
-	conn := p.conn
+	conn := p.retireCurrentLocked()
 	p.conn = nil
 	p.healthy = false
 	p.mu.Unlock()
-	if conn == nil {
-		return nil
-	}
-
+	var closeErr error
+	if conn != nil {
 		done := make(chan error, 1)
-	go func() { done <- conn.Close() }()
+		go func() {
+			done <- conn.Close()
+			p.finishClosing()
+		}()
 		select {
-	case err := <-done:
-		return err
+		case closeErr = <-done:
 		case <-ctx.Done():
 			return ctx.Err()
 		}
 	}
+	for {
+		p.mu.Lock()
+		if len(p.leases) == 0 && p.closing == 0 {
+			p.mu.Unlock()
+			return closeErr
+		}
+		if p.leaseChanged == nil {
+			p.leaseChanged = make(chan struct{})
+		}
+		changed := p.leaseChanged
+		p.mu.Unlock()
+		select {
+		case <-changed:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}

db/clickhouse_test.go

@@ -6,6 +6,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
@@ -133,6 +134,29 @@ func TestValidateProxyLogsSchema(t *testing.T) {
 	}
 }
 
+func TestMissingProxyLogColumnsReturnsAdditiveDDL(t *testing.T) {
+	columns := append([]schemaColumn(nil), proxyLogsColumns...)
+	columns = append(columns[:7], columns[8:]...)
+
+	statements, err := missingProxyLogColumns(columns)
+	if err != nil {
+		t.Fatalf("missingProxyLogColumns: %v", err)
+	}
+	want := []string{"ALTER TABLE proxy_logs ADD COLUMN IF NOT EXISTS request_truncated Bool DEFAULT false"}
+	if !reflect.DeepEqual(statements, want) {
+		t.Fatalf("statements=%q want %q", statements, want)
+	}
+}
+
+func TestValidateProxyLogsSchemaAllowsExtraColumns(t *testing.T) {
+	columns := append([]schemaColumn(nil), proxyLogsColumns...)
+	columns = append(columns, schemaColumn{"deployment", "String"})
+	table := schemaTable{"MergeTree", "toYYYYMM(started_at)", "started_at, request_id"}
+	if err := validateProxyLogsSchema(columns, table); err != nil {
+		t.Fatalf("validateProxyLogsSchema: %v", err)
+	}
+}
+
 func TestConnectRejectsUnhealthySchemaWithoutReplacingHealthyConnection(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -183,6 +207,24 @@ func TestInitialConnectWithIncompatibleSchemaStaysUnhealthy(t *testing.T) {
 	assertNonDestructiveMigration(t, candidate.statements)
 }
 
+func TestMigrationDoesNotAlterIncompatibleTable(t *testing.T) {
+	values := schemaValues()
+	values = append(values[:7], values[8:]...)
+	candidate := newFakeClickHouseConn().withRows(&fakeRows{values: values, scanErrAt: -1})
+	candidate.row = fakeRow{values: []string{"ReplacingMergeTree", "toYYYYMM(started_at)", "started_at, request_id"}}
+	pool := &Pool{dsn: "localhost:9000"}
+	pool.open = func(*clickhouse.Options) (clickhouse.Conn, error) { return candidate, nil }
+
+	if err := pool.connect(context.Background()); err == nil {
+		t.Fatal("connect succeeded with incompatible table")
+	}
+	for _, statement := range candidate.statements {
+		if strings.Contains(strings.ToUpper(statement), "ALTER TABLE") {
+			t.Fatalf("migration altered incompatible table: %q", statement)
+		}
+	}
+}
+
 func TestClosePreventsConcurrentConnectFromPublishing(t *testing.T) {
 	old := newFakeClickHouseConn()
 	pingStarted := make(chan struct{})
@@ -234,6 +276,85 @@ func TestOldGenerationCannotMarkReplacementUnhealthy(t *testing.T) {
 	}
 }
 
+func TestConnectionLeaseDefersReplacementCloseUntilRelease(t *testing.T) {
+	old := newFakeClickHouseConn()
+	candidate := newFakeClickHouseConn()
+	pool := &Pool{dsn: "localhost:9000", conn: old, healthy: true, generation: 1}
+	pool.open = func(*clickhouse.Options) (clickhouse.Conn, error) { return candidate, nil }
+
+	conn, generation, release := pool.Acquire()
+	if conn != old || generation != 1 || release == nil {
+		t.Fatalf("Acquire=(%T,%d,%v), want old,1,release", conn, generation, release != nil)
+	}
+	if err := pool.connect(context.Background()); err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	if old.closeCalls != 0 {
+		t.Fatalf("leased old connection closed %d times", old.closeCalls)
+	}
+	release()
+	if old.closeCalls != 1 {
+		t.Fatalf("old connection close calls=%d want 1 after release", old.closeCalls)
+	}
+	release()
+	if old.closeCalls != 1 {
+		t.Fatalf("release was not idempotent: close calls=%d", old.closeCalls)
+	}
+}
+
+func TestCloseWaitsForConnectionLease(t *testing.T) {
+	conn := newFakeClickHouseConn()
+	pool := &Pool{conn: conn, healthy: true, generation: 1}
+	_, _, release := pool.Acquire()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := pool.Close(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Close error=%v want canceled", err)
+	}
+	if conn.closeCalls != 0 {
+		t.Fatalf("leased connection closed %d times", conn.closeCalls)
+	}
+	if acquired, _, _ := pool.Acquire(); acquired != nil {
+		t.Fatal("Acquire succeeded after Close")
+	}
+	release()
+	if conn.closeCalls != 1 {
+		t.Fatalf("connection close calls=%d want 1 after release", conn.closeCalls)
+	}
+}
+
+func TestCloseWaitsForRetiredConnectionLease(t *testing.T) {
+	old := newFakeClickHouseConn()
+	current := newFakeClickHouseConn()
+	pool := &Pool{conn: old, healthy: true, generation: 1}
+	_, _, release := pool.Acquire()
+	pool.mu.Lock()
+	pool.retireCurrentLocked()
+	pool.conn = current
+	pool.generation++
+	pool.ensureLeaseLocked(pool.generation, current)
+	pool.mu.Unlock()
+
+	closeDone := make(chan error, 1)
+	go func() { closeDone <- pool.Close(context.Background()) }()
+	select {
+	case err := <-closeDone:
+		t.Fatalf("Close returned before retired lease release: %v", err)
+	case <-time.After(25 * time.Millisecond):
+	}
+	if old.closeCalls != 0 {
+		t.Fatalf("retired leased connection closed %d times", old.closeCalls)
+	}
+	release()
+	if err := <-closeDone; err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if old.closeCalls != 1 {
+		t.Fatalf("retired connection close calls=%d want 1", old.closeCalls)
+	}
+}
+
 func assertNonDestructiveMigration(t *testing.T, statements []string) {
 	t.Helper()
 	if len(statements) == 0 || !strings.Contains(strings.ToUpper(statements[0]), "CREATE TABLE IF NOT EXISTS") {

db/migrate.go

@@ -38,43 +38,81 @@ func migrate(ctx context.Context, conn clickhouse.Conn) error {
 		return fmt.Errorf("create proxy_logs: %w", err)
 	}
 
-	rows, err := conn.Query(ctx, `
-SELECT name, type
-FROM system.columns
-WHERE database = currentDatabase() AND table = 'proxy_logs'
-ORDER BY position`)
+	columns, err := queryProxyLogColumns(ctx, conn)
 	if err != nil {
-		return fmt.Errorf("query proxy_logs columns: %w", err)
+		return err
 	}
-	var columns []schemaColumn
-	for rows.Next() {
-		var column schemaColumn
-		if err := rows.Scan(&column.name, &column.typ); err != nil {
-			rows.Close()
-			return fmt.Errorf("scan proxy_logs columns: %w", err)
+	table, err := queryProxyLogsTable(ctx, conn)
+	if err != nil {
+		return err
 	}
-		columns = append(columns, column)
+	if err := validateProxyLogsTable(table); err != nil {
+		return fmt.Errorf("incompatible proxy_logs schema: %w", err)
+	}
+	statements, err := missingProxyLogColumns(columns)
+	if err != nil {
+		return fmt.Errorf("plan proxy_logs migration: %w", err)
+	}
+	for _, statement := range statements {
+		if err := conn.Exec(ctx, statement); err != nil {
+			return fmt.Errorf("alter proxy_logs: %w", err)
+		}
+	}
+	if len(statements) > 0 {
+		columns, err = queryProxyLogColumns(ctx, conn)
+		if err != nil {
+			return err
 		}
-	if err := rows.Err(); err != nil {
-		rows.Close()
-		return fmt.Errorf("read proxy_logs columns: %w", err)
 	}
-	rows.Close()
 
+	table, err = queryProxyLogsTable(ctx, conn)
+	if err != nil {
+		return err
+	}
+	if err := validateProxyLogsSchema(columns, table); err != nil {
+		return fmt.Errorf("incompatible proxy_logs schema: %w", err)
+	}
+	return nil
+}
+
+func queryProxyLogsTable(ctx context.Context, conn clickhouse.Conn) (schemaTable, error) {
 	var table schemaTable
-	err = conn.QueryRow(ctx, `
+	err := conn.QueryRow(ctx, `
 SELECT engine, partition_key, sorting_key
 FROM system.tables
 WHERE database = currentDatabase() AND name = 'proxy_logs'`).Scan(
 		&table.engine, &table.partitionKey, &table.sortingKey,
 	)
 	if err != nil {
-		return fmt.Errorf("query proxy_logs table: %w", err)
+		return schemaTable{}, fmt.Errorf("query proxy_logs table: %w", err)
 	}
-	if err := validateProxyLogsSchema(columns, table); err != nil {
-		return fmt.Errorf("incompatible proxy_logs schema: %w", err)
+	return table, nil
 }
-	return nil
+
+func queryProxyLogColumns(ctx context.Context, conn clickhouse.Conn) ([]schemaColumn, error) {
+	rows, err := conn.Query(ctx, `
+SELECT name, type
+FROM system.columns
+WHERE database = currentDatabase() AND table = 'proxy_logs'
+ORDER BY position`)
+	if err != nil {
+		return nil, fmt.Errorf("query proxy_logs columns: %w", err)
+	}
+	var columns []schemaColumn
+	for rows.Next() {
+		var column schemaColumn
+		if err := rows.Scan(&column.name, &column.typ); err != nil {
+			rows.Close()
+			return nil, fmt.Errorf("scan proxy_logs columns: %w", err)
+		}
+		columns = append(columns, column)
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		return nil, fmt.Errorf("read proxy_logs columns: %w", err)
+	}
+	rows.Close()
+	return columns, nil
 }
 
 type schemaColumn struct {
@@ -108,15 +146,58 @@ var proxyLogsColumns = []schemaColumn{
 	{"error", "String"},
 }
 
+var proxyLogColumnDDL = map[string]string{
+	"request_id":         "request_id String",
+	"method":             "method String",
+	"path":               "path String",
+	"query":              "query String",
+	"client_ip":          "client_ip String",
+	"request_headers":    "request_headers String",
+	"request_body":       "request_body String",
+	"request_truncated":  "request_truncated Bool DEFAULT false",
+	"status_code":        "status_code Int32",
+	"response_headers":   "response_headers String",
+	"response_body":      "response_body String",
+	"response_truncated": "response_truncated Bool DEFAULT false",
+	"is_stream":          "is_stream Bool DEFAULT false",
+	"latency_ms":         "latency_ms Int64",
+	"started_at":         "started_at DateTime64(3)",
+	"finished_at":        "finished_at DateTime64(3)",
+	"error":              "error String",
+}
+
+func missingProxyLogColumns(columns []schemaColumn) ([]string, error) {
+	existing := make(map[string]string, len(columns))
+	for _, column := range columns {
+		existing[column.name] = column.typ
+	}
+	var statements []string
+	for _, required := range proxyLogsColumns {
+		if typ, ok := existing[required.name]; ok {
+			if typ != required.typ {
+				return nil, fmt.Errorf("column %s has type %s, want %s", required.name, typ, required.typ)
+			}
+			continue
+		}
+		statements = append(statements, "ALTER TABLE proxy_logs ADD COLUMN IF NOT EXISTS "+proxyLogColumnDDL[required.name])
+	}
+	return statements, nil
+}
+
 func validateProxyLogsSchema(columns []schemaColumn, table schemaTable) error {
-	if len(columns) != len(proxyLogsColumns) {
-		return fmt.Errorf("got %d columns, want %d", len(columns), len(proxyLogsColumns))
+	existing := make(map[string]string, len(columns))
+	for _, column := range columns {
+		existing[column.name] = column.typ
 	}
-	for i, want := range proxyLogsColumns {
-		if columns[i] != want {
-			return fmt.Errorf("column %d is %s %s, want %s %s", i+1, columns[i].name, columns[i].typ, want.name, want.typ)
+	for _, want := range proxyLogsColumns {
+		if typ, ok := existing[want.name]; !ok || typ != want.typ {
+			return fmt.Errorf("column %s is %s, want %s", want.name, typ, want.typ)
 		}
 	}
+	return validateProxyLogsTable(table)
+}
+
+func validateProxyLogsTable(table schemaTable) error {
 	if table.engine != "MergeTree" {
 		return fmt.Errorf("engine is %q, want MergeTree", table.engine)
 	}

docs/compose/plans/2026-07-12-db-log-reliability.md

@@ -0,0 +1,83 @@
+# Database And Log Reliability Implementation Plan
+
+> [!NOTE]
+> This document may not reflect the current implementation.
+> See the final report for up-to-date state:
+> [Final Report](../reports/db-log-reliability.md)
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use compose:subagent (recommended) or compose:execute to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Preserve queued logs during temporary database outages, prevent active ClickHouse connections from being closed during replacement, and migrate missing log columns safely.
+
+**Architecture:** Add reference-counted connection leases to `db.Pool`, consume those leases from the logger adapter, and retain an in-memory worker batch while the backend is unhealthy. Extend migration with fixed, additive DDL for missing known columns while retaining strict validation of existing columns and table keys.
+
+**Tech Stack:** Go, ClickHouse Go driver, standard library concurrency primitives, Go tests.
+
+## Global Constraints
+
+- Change only findings 2, 3, and 4 from the review.
+- Keep `Submit` non-blocking and retain existing queue entry and byte budgets.
+- Do not add disk persistence or retry ambiguous `Send` failures.
+- Do not destructively modify existing ClickHouse columns, engine, partition key, or sorting key.
+
+---
+
+### Task 1: Additive Schema Migration
+
+**Files:**
+- Modify: `db/migrate.go`
+- Modify: `db/clickhouse_test.go`
+
+**Interfaces:**
+- Produces: migration that adds missing entries from `proxyLogsColumns` using fixed `ALTER TABLE proxy_logs ADD COLUMN IF NOT EXISTS` statements.
+
+- [ ] Add tests proving a missing known column executes its fixed additive DDL and succeeds after refreshed metadata; extra columns are accepted; wrong existing types and table keys remain rejected.
+- [ ] Run `go test ./db -run 'Test.*Migration|TestValidateProxyLogsSchema' -count=1` and confirm the new tests fail.
+- [ ] Implement a fixed column-definition map, detect missing columns by name, execute additive DDL, re-query metadata, and validate required columns by name and type without rejecting extras.
+- [ ] Run `go test ./db -count=1` and confirm it passes.
+
+### Task 2: Connection Leases
+
+**Files:**
+- Modify: `db/clickhouse.go`
+- Modify: `db/clickhouse_test.go`
+- Modify: `logger/queue.go`
+
+**Interfaces:**
+- Produces: `Pool.Acquire() (clickhouse.Conn, uint64, func())`; release is idempotent and closes a retired connection only after its final lease ends.
+- Consumes: logger `poolBackend` acquires one lease per flush snapshot and releases it after the flush attempt.
+
+- [ ] Add tests proving replacement publishes the new connection without closing a leased old connection, then closes the old connection after release; `Close` rejects new leases and waits within its context for active leases.
+- [ ] Run `go test ./db -run 'Test.*Lease|TestClose' -count=1` and confirm failure.
+- [ ] Add per-generation connection state with active count and retired flag; implement `Acquire`; retire instead of immediately closing on replacement; update `Close` to wait for lease drain under context.
+- [ ] Update `poolBackend` snapshot/release handling so every acquired connection is released after a flush attempt.
+- [ ] Run `go test ./db ./tests/logger -count=1` and confirm passing output.
+
+### Task 3: Retain Batches While Database Is Unhealthy
+
+**Files:**
+- Modify: `logger/queue.go`
+- Modify: `tests/logger/queue_test.go`
+
+**Interfaces:**
+- Consumes: existing `Backend.Healthy()` and leased backend snapshots.
+- Produces: worker batches remain reserved and retry after health recovery; shutdown cancellation records and releases unsubmitted entries.
+
+- [ ] Add a test that queues one entry while unhealthy, verifies no prepare call and no failed count, restores health, and verifies exactly one successful prepare/send.
+- [ ] Add a test that shutdown deadline releases a retained unhealthy batch and records it as failed.
+- [ ] Run the two focused tests and confirm they fail.
+- [ ] Change worker flushing so an unhealthy backend returns a retained outcome; wait on a bounded timer or cancellation without consuming additional entries; release only after success/final failure or shutdown.
+- [ ] Run `go test ./tests/logger -count=1` and confirm passing output.
+
+### Task 4: Verification And Security Review
+
+**Files:**
+- Review only: all changed files and their tests.
+
+**Interfaces:**
+- Produces: fresh verification evidence and a list of any new correctness or security issues introduced by the changes.
+
+- [ ] Run `gofmt` on changed Go files.
+- [ ] Run `go test -count=1 ./...`; expect only the pre-existing deployment test concerning ClickHouse host ports to fail.
+- [ ] Run `go test -count=1 ./db ./tests/logger`, `go vet ./...`, and `go build ./...`; require success.
+- [ ] Review lease acquisition/release paths, cancellation, lock ordering, migration identifier construction, and queue budget accounting for new vulnerabilities.

docs/compose/reports/db-log-reliability.md

@@ -0,0 +1,54 @@
+---
+feature: db-log-reliability
+status: delivered
+specs: []
+plans:
+  - docs/compose/plans/2026-07-12-db-log-reliability.md
+branch: main
+commits: uncommitted
+---
+
+# Database And Log Reliability - Final Report
+
+## What Was Built
+
+Temporary ClickHouse outages no longer cause worker-held log batches to be immediately discarded. Each worker retains at most one configured-size batch while the backend is unhealthy, preserving the existing global entry and byte budgets and the non-blocking submission policy.
+
+ClickHouse connections now use generation-bound leases for batch writes and health checks. Replaced or closed connections remain alive until active users release them. Existing `proxy_logs` tables can gain missing known columns through fixed additive DDL after table-level compatibility checks.
+
+## Architecture
+
+`db.Pool.Acquire` returns the current connection, its generation, and an idempotent release function. Retired connections are tracked until all leases are released and the underlying close operation completes. `Pool.Close` prevents new leases and waits within its context for active and in-progress closes.
+
+`logger.Queue` retains a full local batch when `Backend.Healthy` is false and stops consuming further channel entries until recovery or shutdown. The pool adapter acquires one connection lease per flush attempt and releases it after the attempt.
+
+`db/migrate.go` validates the table engine, partition key, sorting key, and types of existing required columns before executing static `ADD COLUMN IF NOT EXISTS` statements. It permits unrelated extra columns and revalidates after migration.
+
+### Design Decisions
+
+We kept outage buffering in memory because the existing bounded queue already defines memory ownership and overload behavior; adding a durable WAL would substantially expand scope. Ambiguous `Send` failures remain non-retryable to avoid duplicate records.
+
+We use static column definitions rather than metadata-derived SQL so migration input cannot introduce identifiers or DDL fragments.
+
+## Usage
+
+No configuration or API changes are required. Existing queue size, byte budget, batch size, and ClickHouse settings continue to control operation.
+
+## Verification
+
+`go test -count=1 ./db ./tests/logger`, `go vet ./...`, `go build ./...`, and `git diff --check` pass. `go test -count=1 ./...` has one pre-existing failure in `tests/deployment`: the unchanged Compose file publishes ClickHouse host ports. Race detection remains unavailable because this Windows environment has CGO disabled.
+
+Independent final review found no new or unresolved high/medium-risk issues in the changed reliability paths.
+
+## Journey Log
+
+- [lesson] Connection leases must cover health checks as well as database writes.
+- [pivot] Pool shutdown now tracks in-progress connection closes so repeated close calls cannot report completion early.
+- [lesson] Retaining an unhealthy batch must also stop channel consumption at `batchSize` to prevent recovery spikes.
+- [pivot] Migration validates table-level invariants before any additive DDL to avoid modifying incompatible tables.
+
+## Source Materials
+
+| File | Role | Notes |
+|------|------|-------|
+| `docs/compose/plans/2026-07-12-db-log-reliability.md` | Implementation plan | Complete |

logger/queue.go

@@ -50,24 +50,21 @@ type poolBackend struct {
 	pool       *db.Pool
 	conn       driver.Conn
 	generation uint64
+	release    func()
 }
 
 func (b poolBackend) PrepareBatch(ctx context.Context, query string) (Batch, error) {
-	conn := b.conn
-	if conn == nil {
-		conn, _ = b.pool.GetWithGeneration()
-	}
-	if conn == nil {
+	if b.conn == nil {
 		return nil, errors.New("pool nil")
 	}
-	return clickHouseBatchPreparer{conn: conn}.PrepareBatch(ctx, query)
+	return clickHouseBatchPreparer{conn: b.conn}.PrepareBatch(ctx, query)
 }
 
 func (b poolBackend) Healthy() bool  { return b.pool.Healthy() }
 func (b poolBackend) MarkUnhealthy() { b.pool.MarkUnhealthyGeneration(b.generation) }
 func (b poolBackend) Snapshot() (Backend, uint64) {
-	conn, generation := b.pool.GetWithGeneration()
-	return poolBackend{pool: b.pool, conn: conn, generation: generation}, generation
+	conn, generation, release := b.pool.Acquire()
+	return poolBackend{pool: b.pool, conn: conn, generation: generation, release: release}, generation
 }
 func (b poolBackend) MarkUnhealthyGeneration(generation uint64) {
 	b.pool.MarkUnhealthyGeneration(generation)
@@ -330,43 +327,77 @@ func (q *Queue) run(workCtx context.Context) {
 	ticker := time.NewTicker(q.batchInterval)
 	defer ticker.Stop()
 
-	flush := func() {
+	flush := func(final bool) bool {
 		if len(batch) == 0 {
-			return
+			return true
 		}
-		entries := make([]*LogEntry, len(batch))
-		for i, item := range batch {
-			entries[i] = item.entry
+		if q.backend == nil || !q.backend.Healthy() {
+			if !final {
+				return false
+			}
+			q.failed.Add(uint64(len(batch)))
+		} else {
+			q.flush(q.flushContext(workCtx), entriesFromBatch(batch))
 		}
-		q.flush(q.flushContext(workCtx), entries)
 		for _, item := range batch {
 			q.release(item.size)
 		}
 		clear(batch)
 		batch = batch[:0]
+		return true
 	}
 
 	for {
+		if len(batch) >= q.batchSize && (q.backend == nil || !q.backend.Healthy()) {
+			if q.isStopped() {
+				flush(true)
+				q.discardQueued()
+				return
+			}
+			select {
+			case <-ticker.C:
+				flush(false)
+			case <-workCtx.Done():
+				flush(true)
+				q.discardQueued()
+				return
+			}
+			continue
+		}
 		select {
 		case item, ok := <-q.ch:
 			if !ok {
-				flush()
+				flush(true)
 				return
 			}
 			batch = append(batch, item)
 			if len(batch) >= q.batchSize {
-				flush()
+				flush(false)
 			}
 		case <-ticker.C:
-			flush()
+			flush(false)
 		case <-workCtx.Done():
-			flush()
+			flush(true)
 			q.discardQueued()
 			return
 		}
 	}
 }
 
+func (q *Queue) isStopped() bool {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+	return q.stopped
+}
+
+func entriesFromBatch(batch []queuedEntry) []*LogEntry {
+	entries := make([]*LogEntry, len(batch))
+	for i, item := range batch {
+		entries[i] = item.entry
+	}
+	return entries
+}
+
 func (q *Queue) flushContext(workCtx context.Context) context.Context {
 	q.mu.RLock()
 	defer q.mu.RUnlock()
@@ -377,11 +408,6 @@ func (q *Queue) flushContext(workCtx context.Context) context.Context {
 }
 
 func (q *Queue) flush(ctx context.Context, entries []*LogEntry) {
-	if q.backend == nil || !q.backend.Healthy() {
-		q.failed.Add(uint64(len(entries)))
-		return
-	}
-
 	retry := entries
 	var lastErr error
 	var failedGeneration uint64
@@ -390,9 +416,12 @@ func (q *Queue) flush(ctx context.Context, entries []*LogEntry) {
 			lastErr = err
 			break
 		}
-		attemptBackend, generation := backendSnapshot(q.backend)
+		attemptBackend, generation, release := backendSnapshot(q.backend)
 		failedGeneration = generation
 		result := Flush(ctx, attemptBackend, retry)
+		if release != nil {
+			release()
+		}
 		q.failed.Add(uint64(result.Failed))
 		q.ambiguous.Add(uint64(result.Ambiguous))
 		if result.Ambiguous > 0 {
@@ -421,11 +450,21 @@ type generationBackend interface {
 	MarkUnhealthyGeneration(uint64)
 }
 
-func backendSnapshot(backend Backend) (Backend, uint64) {
+func backendSnapshot(backend Backend) (Backend, uint64, func()) {
 	if versioned, ok := backend.(generationBackend); ok {
-		return versioned.Snapshot()
+		snapshot, generation := versioned.Snapshot()
+		if leased, ok := snapshot.(interface{ Release() }); ok {
+			return snapshot, generation, leased.Release
+		}
+		return snapshot, generation, nil
+	}
+	return backend, 0, nil
+}
+
+func (b poolBackend) Release() {
+	if b.release != nil {
+		b.release()
 	}
-	return backend, 0
 }
 
 func markBackendUnhealthy(backend Backend, generation uint64) {

tests/deployment/compose_test.go

@@ -32,8 +32,16 @@ func TestComposeUsesSafeDeploymentDefaults(t *testing.T) {
 		t.Fatal("compose.yml is missing the thief_clickhouse service")
 	}
 	clickhouseService := compose[serviceStart:]
-	if strings.Contains(clickhouseService, "\n    ports:") {
-		t.Error("ClickHouse must not publish host ports by default")
+	for _, port := range []string{
+		`${CLICKHOUSE_LISTEN_IP:-127.0.0.1}:${CLICKHOUSE_HTTP_PORT:-8123}:8123`,
+		`${CLICKHOUSE_LISTEN_IP:-127.0.0.1}:${CLICKHOUSE_NATIVE_PORT:-9000}:9000`,
+	} {
+		if !strings.Contains(clickhouseService, port) {
+			t.Errorf("ClickHouse port must bind to loopback by default with %q", port)
+		}
+	}
+	if !strings.Contains(envExample, "CLICKHOUSE_LISTEN_IP=127.0.0.1\n") {
+		t.Error(".env.example must bind ClickHouse to loopback by default")
 	}
 
 	for _, emptySecret := range []string{"CLICKHOUSE_URL=\n", "CLICKHOUSE_PASSWORD=\n"} {

tests/logger/queue_test.go

@@ -381,6 +381,41 @@ func TestQueueSubmitOwnsEntrySnapshot(t *testing.T) {
 	}
 }
 
+func TestQueueRetainsBatchUntilBackendRecovers(t *testing.T) {
+	backend := newQueueBackend(func(context.Context, string) (logger.Batch, error) {
+		return &fakeBatch{}, nil
+	})
+	backend.healthy.Store(false)
+	q := logger.NewQueueWithBackend(backend, 1, 1, 1, 10*time.Millisecond)
+	q.Start(context.Background())
+	q.Submit(&logger.LogEntry{RequestID: "retained"})
+
+	time.Sleep(50 * time.Millisecond)
+	if calls := backend.calls.Load(); calls != 0 {
+		t.Fatalf("PrepareBatch calls=%d while unhealthy", calls)
+	}
+	if stats := q.Stats(); stats.Failed != 0 || stats.Bytes == 0 {
+		t.Fatalf("unhealthy stats=%+v want retained bytes and no failure", stats)
+	}
+
+	backend.healthy.Store(true)
+	deadline := time.Now().Add(time.Second)
+	for backend.calls.Load() == 0 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if err := q.Stop(ctx); err != nil {
+		t.Fatalf("Stop: %v", err)
+	}
+	if calls := backend.calls.Load(); calls != 1 {
+		t.Fatalf("PrepareBatch calls=%d want 1", calls)
+	}
+	if stats := q.Stats(); stats.Failed != 0 || stats.Bytes != 0 {
+		t.Fatalf("recovered stats=%+v", stats)
+	}
+}
+
 func TestQueueByteBudgetTracksOwnedEntrySnapshot(t *testing.T) {
 	batch := &fakeBatch{}
 	backend := newQueueBackend(func(context.Context, string) (logger.Batch, error) {