refactor: introduce Instance connection lifecycle service

internal/instance/errors.go

@@ -0,0 +1,27 @@
+/*
+Copyright 2026.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package instance
+
+// Failure contains only a safe operation message and a Condition reason.
+// Raw driver errors can contain credentials and must not escape through wrapping.
+type Failure struct {
+	Reason    string
+	Operation string
+}
+
+func (e Failure) Error() string           { return e.Operation }
+func (e Failure) ConditionReason() string { return e.Reason }

internal/instance/service.go

@@ -0,0 +1,152 @@
+/*
+Copyright 2026.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package instance manages the lifetime and capabilities of external PostgreSQL instances.
+package instance
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// Credentials are supplied by the credential source, never by the database adapter.
+type Credentials struct{ Username, Password string }
+
+func (Credentials) String() string   { return "[redacted credentials]" }
+func (Credentials) GoString() string { return "[redacted credentials]" }
+
+// CredentialSource resolves administrative credentials when an Instance is assembled.
+type CredentialSource interface {
+	Read(context.Context, api.OpenBaoSecretReference) (Credentials, error)
+}
+
+// Database exposes the capabilities of an assembled Instance.
+type Database interface {
+	Version(context.Context) (string, error)
+	EnsureRegistry(context.Context) error
+	Close()
+}
+
+// Connector creates a database without knowing where credentials came from.
+type Connector interface {
+	Connect(context.Context, api.PostgreSQLEndpoint, Credentials) (Database, error)
+}
+
+type identity struct {
+	uid       types.UID
+	endpoint  api.PostgreSQLEndpoint
+	reference api.OpenBaoSecretReference
+}
+
+type entry struct {
+	identity    identity
+	credentials *Credentials
+	database    Database
+}
+
+// Service owns Instance connections. Reconciles reuse credentials and pools until
+// the UID, endpoint or credential reference changes; allowlist changes do not rotate them.
+// Operations are serialized to prevent Close racing with an active database operation.
+type Service struct {
+	mu        sync.Mutex
+	source    CredentialSource
+	connector Connector
+	entries   map[string]*entry
+	closed    bool
+}
+
+func NewService(source CredentialSource, connector Connector) *Service {
+	return &Service{source: source, connector: connector, entries: make(map[string]*entry)}
+}
+
+func (s *Service) Validate(ctx context.Context, resource *api.PostgreSQLInstance) (string, error) {
+	return s.observe(ctx, resource, false)
+}
+
+func (s *Service) InitializeRegistry(ctx context.Context, resource *api.PostgreSQLInstance) (string, error) {
+	return s.observe(ctx, resource, true)
+}
+
+func (s *Service) observe(ctx context.Context, resource *api.PostgreSQLInstance, ensureRegistry bool) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return "", errors.New("instance service is closed")
+	}
+	db, err := s.database(ctx, resource)
+	if err != nil {
+		return "", err
+	}
+	if ensureRegistry {
+		if err := db.EnsureRegistry(ctx); err != nil {
+			return "", err
+		}
+	}
+	return db.Version(ctx)
+}
+
+func (s *Service) database(ctx context.Context, resource *api.PostgreSQLInstance) (Database, error) {
+	key := identity{uid: resource.UID, endpoint: resource.Spec.Endpoint, reference: resource.Spec.AdminCredentialRef}
+	current := s.entries[resource.Name]
+	if current == nil || current.identity != key {
+		s.release(resource.Name)
+		current = &entry{identity: key}
+		s.entries[resource.Name] = current
+	}
+	if current.credentials == nil {
+		credentials, err := s.source.Read(ctx, key.reference)
+		if err != nil {
+			return nil, err
+		}
+		current.credentials = &credentials
+	}
+	if current.database == nil {
+		database, err := s.connector.Connect(ctx, key.endpoint, *current.credentials)
+		if err != nil {
+			return nil, err
+		}
+		current.database = database
+	}
+	return current.database, nil
+}
+
+// Forget releases local resources only. It never deletes the external registry.
+func (s *Service) Forget(name string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.release(name)
+}
+
+func (s *Service) release(name string) {
+	if current := s.entries[name]; current != nil && current.database != nil {
+		current.database.Close()
+	}
+	delete(s.entries, name)
+}
+
+// Close is called after manager workers stop, and is safe to repeat.
+func (s *Service) Close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.closed = true
+	for name := range s.entries {
+		s.release(name)
+	}
+}

internal/instance/service_test.go

@@ -0,0 +1,156 @@
+/*
+Copyright 2026.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package instance
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+
+	api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type sourceStub struct {
+	reads int
+	err   error
+}
+
+func (s *sourceStub) Read(context.Context, api.OpenBaoSecretReference) (Credentials, error) {
+	s.reads++
+	return Credentials{Username: "admin", Password: "canary-secret"}, s.err
+}
+
+type databaseStub struct {
+	closes, migrations int
+	err                error
+}
+
+func (d *databaseStub) Version(context.Context) (string, error) { return "17", d.err }
+func (d *databaseStub) EnsureRegistry(context.Context) error    { d.migrations++; return d.err }
+func (d *databaseStub) Close()                                  { d.closes++ }
+
+type connectorStub struct {
+	databases []*databaseStub
+	err       error
+}
+
+func (c *connectorStub) Connect(context.Context, api.PostgreSQLEndpoint, Credentials) (Database, error) {
+	if c.err != nil {
+		return nil, c.err
+	}
+	database := &databaseStub{}
+	c.databases = append(c.databases, database)
+	return database, nil
+}
+
+func TestInstanceConnectionLifetime(t *testing.T) {
+	source, connector := &sourceStub{}, &connectorStub{}
+	service := NewService(source, connector)
+	defer service.Close()
+	resource := &api.PostgreSQLInstance{ObjectMeta: metav1.ObjectMeta{Name: "shared", UID: "uid-1"}}
+	ctx := context.Background()
+	check := func() {
+		t.Helper()
+		if _, err := service.Validate(ctx, resource); err != nil {
+			t.Fatalf("validate: %v", err)
+		}
+	}
+	check()
+	if _, err := service.InitializeRegistry(ctx, resource); err != nil {
+		t.Fatal(err)
+	}
+	resource.Generation++
+	resource.Spec.AllowedExtensions = []string{"pg_trgm"}
+	check()
+	if source.reads != 1 || len(connector.databases) != 1 || connector.databases[0].migrations != 1 {
+		t.Fatal("reconciliation or allowlist update unnecessarily reassembled dependencies")
+	}
+	connector.databases[0].err = errors.New("network outage")
+	if _, err := service.Validate(ctx, resource); err == nil {
+		t.Fatal("outage went unnoticed")
+	}
+	connector.databases[0].err = nil
+	check()
+	if source.reads != 1 {
+		t.Fatal("network recovery reread the password")
+	}
+
+	changes := []func(){
+		func() { resource.Spec.Endpoint.Host = "new.example" },
+		func() { resource.Spec.AdminCredentialRef.Path = "new/admin" },
+		func() { resource.UID = "uid-2" },
+	}
+	for index, change := range changes {
+		change()
+		check()
+		if connector.databases[index].closes != 1 {
+			t.Fatal("replaced pool was not closed")
+		}
+	}
+	if source.reads != 4 {
+		t.Fatal("changed connection identity did not reload credentials")
+	}
+	service.Forget(resource.Name)
+	service.Forget(resource.Name)
+	if connector.databases[3].closes != 1 {
+		t.Fatal("deletion did not close pool exactly once")
+	}
+	check()
+	service.Close()
+	service.Close()
+	if connector.databases[4].closes != 1 {
+		t.Fatal("shutdown did not close pool exactly once")
+	}
+	if _, err := service.Validate(ctx, resource); err == nil {
+		t.Fatal("closed service accepted work")
+	}
+}
+
+func TestAssemblyRetriesWithoutPasswordRotation(t *testing.T) {
+	source := &sourceStub{err: errors.New("bao unavailable")}
+	connector := &connectorStub{err: errors.New("connection unavailable")}
+	service := NewService(source, connector)
+	defer service.Close()
+	resource := &api.PostgreSQLInstance{ObjectMeta: metav1.ObjectMeta{Name: "shared"}}
+	ctx := context.Background()
+	if _, err := service.Validate(ctx, resource); err == nil {
+		t.Fatal("expected source failure")
+	}
+	source.err = nil
+	if _, err := service.Validate(ctx, resource); err == nil {
+		t.Fatal("expected connection failure")
+	}
+	connector.err = nil
+	if _, err := service.Validate(ctx, resource); err != nil {
+		t.Fatal(err)
+	}
+	if source.reads != 2 {
+		t.Fatal("retry reread successfully cached credentials")
+	}
+}
+
+func TestCredentialsFormattingIsRedacted(t *testing.T) {
+	value := Credentials{Username: "canary-user", Password: "canary-secret"}
+	for _, format := range []string{"%v", "%+v", "%#v"} {
+		if strings.Contains(fmt.Sprintf(format, value), "canary") {
+			t.Fatal("credential formatting leaked data")
+		}
+	}
+}