mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 12:02:27 +08:00
fix: retry transient database startup errors
This commit is contained in:
@@ -5,7 +5,10 @@ package repository
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/ent"
|
||||
@@ -18,6 +21,72 @@ import (
|
||||
"github.com/lib/pq"
|
||||
)
|
||||
|
||||
const (
|
||||
maxDatabaseInitializationAttempts = 8
|
||||
databaseInitializationRetryBase = time.Second
|
||||
databaseInitializationRetryMax = 30 * time.Second
|
||||
)
|
||||
|
||||
// initializeDatabaseWithRetry retries only errors that indicate PostgreSQL is
|
||||
// temporarily unavailable during startup. Permanent configuration, migration,
|
||||
// and data errors are returned immediately so they remain visible to operators.
|
||||
func initializeDatabaseWithRetry(ctx context.Context, initialize func(context.Context) error) error {
|
||||
return initializeDatabaseWithRetryWithWait(ctx, initialize, waitForDatabaseInitializationRetry)
|
||||
}
|
||||
|
||||
func initializeDatabaseWithRetryWithWait(
|
||||
ctx context.Context,
|
||||
initialize func(context.Context) error,
|
||||
wait func(context.Context, time.Duration) error,
|
||||
) error {
|
||||
var lastErr error
|
||||
for attempt := 1; attempt <= maxDatabaseInitializationAttempts; attempt++ {
|
||||
if err := initialize(ctx); err == nil {
|
||||
return nil
|
||||
} else {
|
||||
lastErr = err
|
||||
if !isTransientDatabaseInitializationError(err) || attempt == maxDatabaseInitializationAttempts {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
delay := databaseInitializationRetryBase * time.Duration(1<<(attempt-1))
|
||||
if delay > databaseInitializationRetryMax {
|
||||
delay = databaseInitializationRetryMax
|
||||
}
|
||||
slog.Warn("database initialization temporarily unavailable; retrying",
|
||||
"attempt", attempt,
|
||||
"max_attempts", maxDatabaseInitializationAttempts,
|
||||
"retry_in", delay,
|
||||
"error", lastErr,
|
||||
)
|
||||
if err := wait(ctx, delay); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return lastErr
|
||||
}
|
||||
|
||||
func waitForDatabaseInitializationRetry(ctx context.Context, delay time.Duration) error {
|
||||
timer := time.NewTimer(delay)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-timer.C:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func isTransientDatabaseInitializationError(err error) bool {
|
||||
var pqErr *pq.Error
|
||||
if !errors.As(err, &pqErr) {
|
||||
return false
|
||||
}
|
||||
code := string(pqErr.Code)
|
||||
return code == "57P03" || strings.HasPrefix(code, "08")
|
||||
}
|
||||
|
||||
// InitEnt 初始化 Ent ORM 客户端并返回客户端实例和底层的 *sql.DB。
|
||||
//
|
||||
// 该函数执行以下操作:
|
||||
@@ -69,7 +138,9 @@ func InitEnt(cfg *config.Config) (*ent.Client, *sql.DB, error) {
|
||||
// 这种方式比 Ent 的自动迁移更可控,支持复杂的迁移场景。
|
||||
migrationCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
|
||||
defer cancel()
|
||||
if err := applyMigrationsFS(migrationCtx, drv.DB(), migrations.FS); err != nil {
|
||||
if err := initializeDatabaseWithRetry(migrationCtx, func(ctx context.Context) error {
|
||||
return applyMigrationsFS(ctx, drv.DB(), migrations.FS)
|
||||
}); err != nil {
|
||||
_ = drv.Close() // 迁移失败时关闭驱动,避免资源泄露
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/lib/pq"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func transientDatabaseError(code string) error {
|
||||
return &pq.Error{Code: pq.ErrorCode(code), Message: "database is temporarily unavailable"}
|
||||
}
|
||||
|
||||
func TestIsTransientDatabaseInitializationError(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{name: "postgres is starting", err: transientDatabaseError("57P03"), want: true},
|
||||
{name: "connection failure", err: fmt.Errorf("wrapped: %w", transientDatabaseError("08006")), want: true},
|
||||
{name: "authentication failure", err: transientDatabaseError("28P01"), want: false},
|
||||
{name: "migration error", err: errors.New("migration checksum mismatch"), want: false},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
require.Equal(t, tt.want, isTransientDatabaseInitializationError(tt.err))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitializeDatabaseWithRetryEventuallySucceeds(t *testing.T) {
|
||||
attempts := 0
|
||||
var delays []time.Duration
|
||||
err := initializeDatabaseWithRetryWithWait(context.Background(), func(context.Context) error {
|
||||
attempts++
|
||||
if attempts <= 3 {
|
||||
return transientDatabaseError("57P03")
|
||||
}
|
||||
return nil
|
||||
}, func(_ context.Context, delay time.Duration) error {
|
||||
delays = append(delays, delay)
|
||||
return nil
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 4, attempts)
|
||||
require.Equal(t, []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}, delays)
|
||||
}
|
||||
|
||||
func TestInitializeDatabaseWithRetryFailsFastForPermanentError(t *testing.T) {
|
||||
attempts := 0
|
||||
waitCalled := false
|
||||
permanentErr := transientDatabaseError("28P01")
|
||||
err := initializeDatabaseWithRetryWithWait(context.Background(), func(context.Context) error {
|
||||
attempts++
|
||||
return permanentErr
|
||||
}, func(_ context.Context, _ time.Duration) error {
|
||||
waitCalled = true
|
||||
return nil
|
||||
})
|
||||
|
||||
require.ErrorIs(t, err, permanentErr)
|
||||
require.Equal(t, 1, attempts)
|
||||
require.False(t, waitCalled)
|
||||
}
|
||||
|
||||
func TestInitializeDatabaseWithRetryStopsWhenContextIsCanceled(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
attempts := 0
|
||||
err := initializeDatabaseWithRetryWithWait(ctx, func(context.Context) error {
|
||||
attempts++
|
||||
return transientDatabaseError("57P03")
|
||||
}, func(ctx context.Context, _ time.Duration) error {
|
||||
cancel()
|
||||
return ctx.Err()
|
||||
})
|
||||
|
||||
require.ErrorIs(t, err, context.Canceled)
|
||||
require.Equal(t, 1, attempts)
|
||||
}
|
||||
|
||||
func TestInitializeDatabaseWithRetryReturnsLastErrorAfterLimit(t *testing.T) {
|
||||
attempts := 0
|
||||
lastErr := transientDatabaseError("08006")
|
||||
var delays []time.Duration
|
||||
err := initializeDatabaseWithRetryWithWait(context.Background(), func(context.Context) error {
|
||||
attempts++
|
||||
return lastErr
|
||||
}, func(_ context.Context, delay time.Duration) error {
|
||||
delays = append(delays, delay)
|
||||
return nil
|
||||
})
|
||||
|
||||
require.ErrorIs(t, err, lastErr)
|
||||
require.Equal(t, maxDatabaseInitializationAttempts, attempts)
|
||||
require.Len(t, delays, maxDatabaseInitializationAttempts-1)
|
||||
require.Equal(t, databaseInitializationRetryMax, delays[len(delays)-1])
|
||||
}
|
||||
|
||||
func TestInitializeDatabaseWithRetryAllowsIdempotentMigrationRetry(t *testing.T) {
|
||||
applied := make(map[string]bool)
|
||||
attempts := 0
|
||||
err := initializeDatabaseWithRetryWithWait(context.Background(), func(context.Context) error {
|
||||
attempts++
|
||||
if !applied["001_init.sql"] {
|
||||
applied["001_init.sql"] = true
|
||||
return transientDatabaseError("57P03")
|
||||
}
|
||||
return nil
|
||||
}, func(_ context.Context, _ time.Duration) error { return nil })
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 2, attempts)
|
||||
require.True(t, applied["001_init.sql"])
|
||||
}
|
||||
@@ -49,6 +49,20 @@ volumes:
|
||||
redis_data:
|
||||
```
|
||||
|
||||
## Startup and Database Recovery
|
||||
|
||||
Sub2API runs database migrations while starting. PostgreSQL may still be
|
||||
recovering briefly after a host or Docker daemon restart. The application
|
||||
retries transient PostgreSQL startup and connection errors with bounded
|
||||
exponential backoff, then continues startup when the database is ready.
|
||||
Permanent errors such as invalid credentials, migration checksum mismatches,
|
||||
SQL errors, and incompatible data fail immediately.
|
||||
|
||||
The Compose deployment also checks PostgreSQL readiness with both `pg_isready`
|
||||
and a simple SQL query. `depends_on: condition: service_healthy` helps order a
|
||||
fresh Compose start, but application-level retries are still required when
|
||||
Docker restores existing containers after a host restart.
|
||||
|
||||
## Environment Variables
|
||||
|
||||
| Variable | Description | Required | Default |
|
||||
|
||||
@@ -146,6 +146,28 @@ When using Docker Compose with `AUTO_SETUP=true`:
|
||||
docker compose logs sub2api | grep "admin password"
|
||||
```
|
||||
|
||||
### Startup and Database Recovery
|
||||
|
||||
Sub2API applies database migrations during application startup. PostgreSQL can
|
||||
remain in its recovery/startup phase briefly after a host or Docker daemon
|
||||
restart. The application retries transient PostgreSQL startup and connection
|
||||
errors with bounded exponential backoff, then starts automatically when the
|
||||
database becomes ready. Authentication errors, migration checksum mismatches,
|
||||
SQL errors, and other permanent configuration or data errors fail immediately.
|
||||
|
||||
The Compose example also uses a PostgreSQL health check that verifies both
|
||||
server readiness and a simple SQL query. `depends_on: condition: service_healthy`
|
||||
controls dependency ordering for a fresh Compose start, but it is not a
|
||||
replacement for application-level retries when Docker restores existing
|
||||
containers after a host restart.
|
||||
|
||||
For systemd deployments, keep `Restart=always` and `RestartSec` configured in
|
||||
`sub2api.service`; the application retry covers transient database startup,
|
||||
while systemd remains the supervisor for permanent process exits. For
|
||||
Kubernetes, use a PostgreSQL readiness probe and retain the Sub2API startup
|
||||
retry behavior; configure the application liveness probe separately so a
|
||||
database recovery period is not treated as a permanent process failure.
|
||||
|
||||
### Database Migration Notes (PostgreSQL)
|
||||
|
||||
- Migrations are applied in lexicographic order (e.g. `001_...sql`, `002_...sql`).
|
||||
|
||||
@@ -221,7 +221,9 @@ services:
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 3
|
||||
start_period: 30s
|
||||
# The application also retries transient database startup errors. This
|
||||
# grace period keeps Compose from declaring it unhealthy during bootstrap.
|
||||
start_period: 120s
|
||||
|
||||
# ===========================================================================
|
||||
# PostgreSQL Database
|
||||
@@ -260,12 +262,12 @@ services:
|
||||
test:
|
||||
[
|
||||
"CMD-SHELL",
|
||||
"pg_isready -U ${POSTGRES_USER:-sub2api} -d ${POSTGRES_DB:-sub2api}",
|
||||
"pg_isready -U ${POSTGRES_USER:-sub2api} -d ${POSTGRES_DB:-sub2api} && psql -U ${POSTGRES_USER:-sub2api} -d ${POSTGRES_DB:-sub2api} -Atqc 'SELECT 1' >/dev/null",
|
||||
]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
retries: 18
|
||||
start_period: 60s
|
||||
# 注意:不暴露端口到宿主机,应用通过内部网络连接
|
||||
# 如需调试,可临时添加:ports: ["127.0.0.1:5433:5432"]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user