fix: prevent reconnect after pool close

db/clickhouse_test.go

@@ -183,6 +183,37 @@ func TestInitialConnectWithIncompatibleSchemaStaysUnhealthy(t *testing.T) {
 	assertNonDestructiveMigration(t, candidate.statements)
 }
 
+func TestClosePreventsConcurrentConnectFromPublishing(t *testing.T) {
+	old := newFakeClickHouseConn()
+	pingStarted := make(chan struct{})
+	pingRelease := make(chan struct{})
+	candidate := newFakeClickHouseConn().withBlockedPing(pingStarted, pingRelease)
+	pool := &Pool{dsn: "localhost:9000", conn: old, healthy: true}
+	pool.open = func(*clickhouse.Options) (clickhouse.Conn, error) { return candidate, nil }
+
+	connectDone := make(chan error, 1)
+	go func() { connectDone <- pool.connect(context.Background()) }()
+	<-pingStarted
+
+	if err := pool.Close(context.Background()); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	close(pingRelease)
+	if err := <-connectDone; err == nil {
+		t.Fatal("connect succeeded after pool was closed")
+	}
+
+	if pool.Get() != nil || pool.Healthy() {
+		t.Fatalf("closed pool connection = %T, healthy = %v; want nil, false", pool.Get(), pool.Healthy())
+	}
+	if old.closeCalls != 1 {
+		t.Fatalf("old connection close calls = %d, want 1", old.closeCalls)
+	}
+	if candidate.closeCalls != 1 {
+		t.Fatalf("candidate close calls = %d, want 1", candidate.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") {
@@ -197,12 +228,14 @@ func assertNonDestructiveMigration(t *testing.T, statements []string) {
 }
 
 type fakeClickHouseConn struct {
-	rows       *fakeRows
-	row        driver.Row
-	execErr    error
-	queryErr   error
-	statements []string
-	closeCalls int
+	rows        *fakeRows
+	row         driver.Row
+	execErr     error
+	queryErr    error
+	statements  []string
+	closeCalls  int
+	pingStarted chan struct{}
+	pingRelease chan struct{}
 }
 
 func newFakeClickHouseConn() *fakeClickHouseConn {
@@ -229,6 +262,11 @@ func (c *fakeClickHouseConn) withColumnType(index int, typ string) *fakeClickHou
 	c.rows.values[index][1] = typ
 	return c
 }
+func (c *fakeClickHouseConn) withBlockedPing(started, release chan struct{}) *fakeClickHouseConn {
+	c.pingStarted = started
+	c.pingRelease = release
+	return c
+}
 
 func (c *fakeClickHouseConn) Contributors() []string                            { return nil }
 func (c *fakeClickHouseConn) ServerVersion() (*driver.ServerVersion, error)     { return nil, nil }
@@ -249,9 +287,15 @@ func (c *fakeClickHouseConn) Exec(_ context.Context, query string, _ ...any) err
 	return c.execErr
 }
 func (c *fakeClickHouseConn) AsyncInsert(context.Context, string, bool, ...any) error { return nil }
-func (c *fakeClickHouseConn) Ping(context.Context) error                              { return nil }
-func (c *fakeClickHouseConn) Stats() driver.Stats                                     { return driver.Stats{} }
-func (c *fakeClickHouseConn) Close() error                                            { c.closeCalls++; return nil }
+func (c *fakeClickHouseConn) Ping(context.Context) error {
+	if c.pingStarted != nil {
+		close(c.pingStarted)
+		<-c.pingRelease
+	}
+	return nil
+}
+func (c *fakeClickHouseConn) Stats() driver.Stats { return driver.Stats{} }
+func (c *fakeClickHouseConn) Close() error        { c.closeCalls++; return nil }
 
 type fakeRows struct {
 	values    [][]string