Go бази на податоци: Архитектирање на трансакциска доверливост со базени, временски ограничувања и безбедни миграции

Go Databases: Architecting Transactional Reliability with Pools, Timeouts, and Safe Migrations

Go бази на податоци: Архитектирање на трансакциска доверливост со базени, временски ограничувања и безбедни миграции

Откажувањето на базата на податоци ретко започнува со драматичен прекин. Почесто, латентноста расте, барањата ги задржуваат врските подолго, пулот се заситува, а повторните обиди го умножуваат оптоварувањето. Трансакција што изгледала совршено исправна во развојната средина станува дел од повратна спрега во продукција.

Овој туторијал создава мала Go услуга за трансфери што префрла целобројни валутни единици меѓу сметки. Користи PostgreSQL трансакции, детерминистичко заклучување на редови, клучеви за идемпотентност, ограничен пул на врски, посебни рокови за врска и за пребарување, и миграции управувани во рамките на изданијата. Резултатот е намерно компактен, но неговите граници на доверливост се реални.

Архитектура и договор за доверливост

HTTP услугата слуша на 127.0.0.1:8080 и се поврзува со PostgreSQL на 127.0.0.1:5433. Трансферот задолжува една сметка, одобрува друга и го запишува својот клуч за идемпотентност во една трансакција.

Важните гаранции се:

Салдата и статусот на трансферот се потврдуваат атомски.
Истовремените трансфери ги заклучуваат сметките по растечки редослед на ID, со што се намалува ризикот од застој.
Повторен клуч за идемпотентност враќа успех само кога неговиот товар се совпаѓа со првичното барање.
Пулот има строго ограничување на големината; стекнувањето и користењето врска се ограничени од контекстот на барањето.
Воспоставувањето врска, извршувањето SQL, стекнувањето заклучување, HTTP читањата и исклучувањето имаат различни буџети.
Миграциите се извршуваат како чекор од изданието, никогаш опортунистички од секоја инстанца на апликацијата.

Ова не прави надворешна уплата или известување точно еднаш. Ако подоцнежен работник испраќа споредни ефекти, мора да користи outbox или друга граница на идемпотентен ефект и да прифати испорака најмалку еднаш.

Предуслови и распоред на проектот

Користете Go 1.22 или понов, Docker или еквивалентна PostgreSQL инсталација, и PostgreSQL клиентски алатки. Единствената зависност на апликацијата е github.com/jackc/pgx/v5 од компатибилната линија v5.7.x. Командите за миграција користат github.com/golang-migrate/migrate/v4 верзија v4.17.1.

ledger/
├── go.mod
├── main.go
└── migrations/
    └── 000001_init.up.sql

Создадете развојна база на податоци. Објавената порта е ограничена на loopback; dev-only намерно не е продукциски акредитив.

docker run --name ledger-db \
  -e POSTGRES_PASSWORD=dev-only \
  -e POSTGRES_DB=ledger \
  -p 127.0.0.1:5433:5432 \
  -d postgres:16

mkdir -p ledger/migrations
cd ledger

go mod init example.com/ledger
go get github.com/jackc/pgx/[email protected]
go install -tags postgres github.com/golang-migrate/migrate/v4/cmd/[email protected]

Создадете шема управувана од ограничувања

Чувајте ги парите како целобројни помали единици, а не како вредности со подвижна запирка. Ограничувањата на базата на податоци остануваат вредни дури и кога Go ги проверува истите услови: тие ја штитат инваријантата од скрипти за одржување, идни услуги и програмерски грешки.

-- migrations/000001_init.up.sql
BEGIN;

CREATE TABLE accounts (
    id         bigint PRIMARY KEY,
    balance    bigint NOT NULL CHECK (balance >= 0),
    updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE transfer_requests (
    idempotency_key text PRIMARY KEY,
    from_id         bigint NOT NULL REFERENCES accounts(id),
    to_id           bigint NOT NULL REFERENCES accounts(id),
    amount           bigint NOT NULL CHECK (amount > 0),
    status           text NOT NULL CHECK (status IN ('pending', 'completed')),
    created_at       timestamptz NOT NULL DEFAULT now(),
    completed_at     timestamptz
);

COMMIT;

Инсталирајте ги миграциите со изрични оперативни ограничувања. Рокот за поврзување го покрива воспоставувањето врска со PostgreSQL; не ги ограничува изјавите за миграција. PostgreSQL-овите lock_timeout и statement_timeout ги обезбедуваат тие одделни граници.

export DATABASE_URL='postgres://postgres:[email protected]:5433/ledger?sslmode=disable'

PGCONNECT_TIMEOUT=3 \
PGOPTIONS='-c lock_timeout=2s -c statement_timeout=15min' \
migrate -path migrations -database "$DATABASE_URL" up

psql "$DATABASE_URL" -v ON_ERROR_STOP=1 <<'SQL'
INSERT INTO accounts (id, balance)
VALUES (1, 10000), (2, 10000);
SQL

Имплементирајте ја трансакцијата и нејзините буџети

Двосекундниот контекст на апликацијата го ограничува чекањето на пулот и извршувањето SQL. Во PostgreSQL, рокот за изјава е малку пократок, додека стекнувањето заклучување добива само 500 милисекунди. Овој редослед остава време да се преведе неуспехот и да се врати одговор.

Ограничувањето на пулот е контрола на повратен притисок, а не цел за перформанси. Поставете го под капацитетот на врските на базата на податоци откако ќе резервирате простор за миграции, администрација, реплики и други услуги.

// main.go
package main

import (
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var (
	errFunds = errors.New("insufficient funds")
	errKey   = errors.New("idempotency key reused with different payload")
	errAcct  = errors.New("account not found")
)

type request struct {
	FromID int64 `json:"from_id"`
	ToID   int64 `json:"to_id"`
	Amount int64 `json:"amount"`
}

type app struct{ db *pgxpool.Pool }

func (a *app) transfer(ctx context.Context, key string, r request) error {
	tx, err := a.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return err
	}
	defer tx.Rollback(context.Background())

	if _, err = tx.Exec(ctx, "SET LOCAL statement_timeout = '1500ms'"); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, "SET LOCAL lock_timeout = '500ms'"); err != nil {
		return err
	}

	tag, err := tx.Exec(ctx, `
		INSERT INTO transfer_requests
		    (idempotency_key, from_id, to_id, amount, status)
		VALUES ($1, $2, $3, $4, 'pending')
		ON CONFLICT DO NOTHING`,
		key, r.FromID, r.ToID, r.Amount)
	if err != nil {
		return err
	}

	if tag.RowsAffected() == 0 {
		var from, to, amount int64
		var status string
		err = tx.QueryRow(ctx, `
			SELECT from_id, to_id, amount, status
			FROM transfer_requests
			WHERE idempotency_key = $1`, key).
			Scan(&from, &to, &amount, &status)
		if err != nil {
			return err
		}
		if from != r.FromID || to != r.ToID || amount != r.Amount {
			return errKey
		}
		if status != "completed" {
			return errors.New("transfer has unexpected state")
		}
		return tx.Commit(ctx)
	}

	rows, err := tx.Query(ctx, `
		SELECT id FROM accounts
		WHERE id IN ($1, $2)
		ORDER BY id
		FOR UPDATE`, r.FromID, r.ToID)
	if err != nil {
		return err
	}
	count := 0
	for rows.Next() {
		count++
	}
	rows.Close()
	if err = rows.Err(); err != nil {
		return err
	}
	if count != 2 {
		return errAcct
	}

	tag, err = tx.Exec(ctx, `
		UPDATE accounts
		SET balance = balance - $1, updated_at = now()
		WHERE id = $2 AND balance >= $1`, r.Amount, r.FromID)
	if err != nil {
		return err
	}
	if tag.RowsAffected() != 1 {
		return errFunds
	}

	if _, err = tx.Exec(ctx, `
		UPDATE accounts
		SET balance = balance + $1, updated_at = now()
		WHERE id = $2`, r.Amount, r.ToID); err != nil {
		return err
	}

	if _, err = tx.Exec(ctx, `
		UPDATE transfer_requests
		SET status = 'completed', completed_at = now()
		WHERE idempotency_key = $1`, key); err != nil {
		return err
	}

	return tx.Commit(ctx)
}

func (a *app) handleTransfer(w http.ResponseWriter, req *http.Request) {
	req.Body = http.MaxBytesReader(w, req.Body, 4096)
	var in request
	dec := json.NewDecoder(req.Body)
	dec.DisallowUnknownFields()
	if dec.Decode(&in) != nil || in.FromID <= 0 || in.ToID <= 0 ||
		in.FromID == in.ToID || in.Amount <= 0 || in.Amount > 1_000_000_000_000 {
		http.Error(w, "invalid request", http.StatusBadRequest)
		return
	}

	key := req.Header.Get("Idempotency-Key")
	if len(key) < 8 || len(key) > 128 {
		http.Error(w, "invalid Idempotency-Key", http.StatusBadRequest)
		return
	}

	ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second)
	defer cancel()
	start := time.Now()
	err := a.transfer(ctx, key, in)
	slog.Info("transfer", "key", key, "duration", time.Since(start), "error", err)

	switch {
	case err == nil:
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]string{"status": "completed"})
	case errors.Is(err, errFunds), errors.Is(err, errKey), errors.Is(err, errAcct):
		http.Error(w, err.Error(), http.StatusConflict)
	case errors.Is(err, context.DeadlineExceeded):
		http.Error(w, "database deadline exceeded", http.StatusGatewayTimeout)
	default:
		http.Error(w, "internal error", http.StatusInternalServerError)
	}
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(),
		syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	cfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
	if err != nil {
		panic(err)
	}
	cfg.MaxConns = 20
	cfg.MinConns = 2
	cfg.MaxConnLifetime = 30 * time.Minute
	cfg.MaxConnIdleTime = 5 * time.Minute
	cfg.HealthCheckPeriod = time.Minute
	cfg.ConnConfig.ConnectTimeout = 3 * time.Second

	db, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		panic(err)
	}
	defer db.Close()

	pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
	err = db.Ping(pingCtx)
	cancel()
	if err != nil {
		panic(err)
	}

	a := &app{db: db}
	mux := http.NewServeMux()
	mux.HandleFunc("POST /transfers", a.handleTransfer)
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
		healthCtx, cancel := context.WithTimeout(r.Context(), 500*time.Millisecond)
		defer cancel()
		if err := db.Ping(healthCtx); err != nil {
			http.Error(w, "unhealthy", http.StatusServiceUnavailable)
			return
		}
		w.WriteHeader(http.StatusNoContent)
	})

	srv := &http.Server{
		Addr:              "127.0.0.1:8080",
		Handler:           mux,
		ReadHeaderTimeout: 2 * time.Second,
		ReadTimeout:       5 * time.Second,
		WriteTimeout:      5 * time.Second,
		IdleTimeout:       60 * time.Second,
	}

	go func() {
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			slog.Error("server stopped", "error", err)
			stop()
		}
	}()

	<-ctx.Done()
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
	defer cancel()
	if err := srv.Shutdown(shutdownCtx); err != nil {
		slog.Error("shutdown", "error", err)
	}
}

Тестирајте успех, повторување и враќање назад

Стартувајте ја услугата, испратете го истото барање двапати, па проверете ги салдата. И двата повика треба да пријават завршување, но промената на салдото мора да се случи еднаш.

DATABASE_URL="$DATABASE_URL" go run .

curl --fail-with-body -X POST http://127.0.0.1:8080/transfers \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: transfer-demo-0001' \
  -d '{"from_id":1,"to_id":2,"amount":1250}'

curl --fail-with-body -X POST http://127.0.0.1:8080/transfers \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: transfer-demo-0001' \
  -d '{"from_id":1,"to_id":2,"amount":1250}'

psql "$DATABASE_URL" -c \
  'SELECT id, balance FROM accounts ORDER BY id;'

Очекуваните салда се 8750 и 11250. Повторно употребете го клучот со друг износ и очекувајте HTTP 409. Побарајте повеќе од достапното салдо и потврдете дека ниту една сметка ниту transfer_requests не задржува делумна промена. За тестирање на конкуренција, извршете истовремени трансфери во двете насоки; детерминистичкото заклучување треба да спречи вообичаени застои, додека рокот за заклучување ги одржува патолошките чекања конечни.

Мигрирајте без претворање на распоредувањето во инцидент со заклучување

За постоечка табела, избегнувајте комбинирање на долго пополнување наназад, ново ограничување и пуштање на апликацијата во една миграција. Користете проширување и стеснување:

Додадете nullable колона или компатибилна структура на табела со краток рок за заклучување.
Распоредете код што ги разбира и старите и новите претставувања.
Пополнете наназад во мали, потврдени серии со курсор што може повторно да се стартува.
Додајте валидација одделно; ограничувањата NOT VALID на PostgreSQL можат да го намалат почетниот прозорец на блокирање каде што е применливо.
Валидирајте, префрлете ги читањата и отстранете ги застарените структури во подоцнежно издание.

Изградете големи индекси со CREATE INDEX CONCURRENTLY во миграција што не е обвиткана со BEGIN. Дајте ѝ сопствен дарежлив, но конечен рок за изјава. Ако migrate пријави валкана верзија, проверете ја базата на податоци и ефектите од миграцијата пред да користите force; присилувањето само ги менува метаподатоците и не поправа половично применет DDL.

Безбедност, набљудливост и перформанси

Продукциските акредитиви треба да припаѓаат на login улога со само потребните привилегии за табели и секвенци. Користете TLS кога базата на податоци преминува преку недоверлива мрежа, вчитувајте тајни од датотека со околински променливи читлива само за root или од менаџер за тајни и никогаш не евидентирајте URL-адреси на базата на податоци или тела на барања. Заштитете го PostgreSQL со заштитен ѕид од јавни интерфејси.

JSON дневникот веќе ги запишува латентноста и класата на неуспех. Додајте метрики за стекнати врски од пулот, неактивни врски, времетраење на стекнувањето, истекувања на рокови за пребарување, истекувања на рокови за заклучување, враќања на трансакции и HTTP статусни кодови. Алармирањето за времето на чекање на пулот често е покорисно отколку следењето само на бројот на активни врски.

Не обидувајте се автоматски повторно за секоја грешка на базата на податоци. Повторен обид е соодветен само за минливи, класифицирани неуспеси како конфликти при серијализација, и му се потребни jitter, мал лимит на обиди и првичниот рок на барањето. Идемпотентноста го прави повторниот обид на целото барање побезбеден; не го прави безопасен притисокот од неограничени повторни обиди.

Распоредете како ограничен систем

Изградете статички бинарен фајл каде што е поддржано, инсталирајте го под изричен директориум за апликацијата и стартувајте го како непривилегиран systemd корисник. Администратор мора да ја создаде сервисната сметка и да го заштити /etc/ledger/ledger.env со режим 0600.

CGO_ENABLED=0 go build -trimpath -o ledger .
sudo install -d -o root -g root -m 0755 /opt/ledger
sudo install -o root -g root -m 0755 ./ledger /opt/ledger/ledger

# Release job: migrate first, then restart only after success.
PGCONNECT_TIMEOUT=3 \
PGOPTIONS='-c lock_timeout=2s -c statement_timeout=15min' \
migrate -path migrations -database "$DATABASE_URL" up

sudo systemctl restart ledger.service
curl --fail http://127.0.0.1:8080/healthz

Поставете зајакнат обратен прокси пред неа за TLS, автентикација, ограничувања на стапката на барања и јавно изложување. Чувајте ја апликацијата врзана за loopback, освен ако мрежниот дизајн изрично не бара поинаку. При постепено распоредување, осигурете се дека дозволеното време за исклучување ги надминува двосекундниот рок за базата на податоци и петсекундниот рок за HTTP запишување.

Конечна листа за проверка

Максимумот на пулот се вклопува во резервираниот буџет за врски на PostgreSQL.
Роковите за поврзување, барање, изјава, заклучување и исклучување се конфигурирани независно.
Редовите на сметките се заклучуваат во детерминистички редослед.
Клучевите за идемпотентност се врзуваат за целосниот товар на операцијата.
Неуспешните трансфери не оставаат задолжување, одобрување или потврден запис во состојба на чекање.
Дупликатните барања ги менуваат салдата само еднаш.
Миграциите се извршуваат еднаш по издание со ограничени заклучувања и проверени неуспеси.
Базата на податоци е приватна, акредитивите се со најмали привилегии, а тајните отсуствуваат од дневниците.
Чекањето на пулот, латентноста на пребарувањата, истекувањата на рокови, враќањата назад и одговорите со грешка се набљудливи.

Доверливиот код за база на податоци не се дефинира само со трансакцијата по среќната патека. Тоа е кореографијата околу таа трансакција: колку долго смее да чека, колку конкуренти можат да се натпреваруваат, што значи повторен обид, како промените на шемата коегзистираат со жив сообраќај и како исклучувањето ја прекинува работата. Направете ги тие граници изрични, и базата на податоци станува контролиран дел од системот наместо неговиот најмистериозен засилувач на неуспеси.