From 6f3716525689000aa35abc5d30e019496e121c33 Mon Sep 17 00:00:00 2001 From: m1saka Date: Mon, 13 Jul 2026 11:27:34 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=97=A5=E5=BF=97=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8C=96=E5=8F=AF=E9=9D=A0=E6=80=A7=E4=B8=8E=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E5=AE=89=E5=85=A8=E9=BB=98=E8=AE=A4=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 2 +- compose.yml | 4 +- db/clickhouse.go | 146 ++++++++++++++++-- db/clickhouse_test.go | 121 +++++++++++++++ db/migrate.go | 137 ++++++++++++---- .../plans/2026-07-12-db-log-reliability.md | 83 ++++++++++ docs/compose/reports/db-log-reliability.md | 54 +++++++ logger/queue.go | 93 +++++++---- tests/deployment/compose_test.go | 12 +- tests/logger/queue_test.go | 35 +++++ 10 files changed, 612 insertions(+), 75 deletions(-) create mode 100644 docs/compose/plans/2026-07-12-db-log-reliability.md create mode 100644 docs/compose/reports/db-log-reliability.md diff --git a/.env.example b/.env.example index 2f03a7f..f3ac132 100644 --- a/.env.example +++ b/.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 diff --git a/compose.yml b/compose.yml index f88c57a..c2cb9fb 100644 --- a/compose.yml +++ b/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}" diff --git a/db/clickhouse.go b/db/clickhouse.go index e49e8df..9bf6b2e 100644 --- a/db/clickhouse.go +++ b/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() + p.finishClosing() + }() + select { + case closeErr = <-done: + case <-ctx.Done(): + return ctx.Err() + } } - - done := make(chan error, 1) - go func() { done <- conn.Close() }() - select { - case err := <-done: - return err - 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() + } } } diff --git a/db/clickhouse_test.go b/db/clickhouse_test.go index 92f73f3..d2fe18d 100644 --- a/db/clickhouse_test.go +++ b/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") { diff --git a/db/migrate.go b/db/migrate.go index abde6a1..12da9b2 100644 --- a/db/migrate.go +++ b/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 + } + 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) } - columns = append(columns, column) } - if err := rows.Err(); err != nil { - rows.Close() - return fmt.Errorf("read proxy_logs columns: %w", err) + if len(statements) > 0 { + columns, err = queryProxyLogColumns(ctx, conn) + if err != nil { + return 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 +} + +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) } - return nil + 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"}, } -func validateProxyLogsSchema(columns []schemaColumn, table schemaTable) error { - if len(columns) != len(proxyLogsColumns) { - return fmt.Errorf("got %d columns, want %d", len(columns), len(proxyLogsColumns)) +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 } - 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) + 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 { + existing := make(map[string]string, len(columns)) + for _, column := range columns { + existing[column.name] = column.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) } diff --git a/docs/compose/plans/2026-07-12-db-log-reliability.md b/docs/compose/plans/2026-07-12-db-log-reliability.md new file mode 100644 index 0000000..51c5120 --- /dev/null +++ b/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. diff --git a/docs/compose/reports/db-log-reliability.md b/docs/compose/reports/db-log-reliability.md new file mode 100644 index 0000000..112870f --- /dev/null +++ b/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 | diff --git a/logger/queue.go b/logger/queue.go index e801230..8498f0f 100644 --- a/logger/queue.go +++ b/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) { diff --git a/tests/deployment/compose_test.go b/tests/deployment/compose_test.go index 2abb132..c75f5b0 100644 --- a/tests/deployment/compose_test.go +++ b/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"} { diff --git a/tests/logger/queue_test.go b/tests/logger/queue_test.go index 99e0fa6..4b64bbf 100644 --- a/tests/logger/queue_test.go +++ b/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) {