This commit is contained in:
@@ -0,0 +1,243 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
applicationledger "gl/application/ledger"
|
||||
applicationsettlement "gl/application/settlement"
|
||||
domain "gl/domain/ledger"
|
||||
ledgerv1 "gl/gen/ledger/v1"
|
||||
grpcadapter "gl/interface/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
// Opt-in real PostgreSQL proof. Each run owns a new schema and removes only it.
|
||||
func TestAuthorityWithPostgres(t *testing.T) {
|
||||
dsn := os.Getenv("GL_TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("set GL_TEST_DATABASE_URL to a disposable PostgreSQL database")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
admin, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer admin.Close()
|
||||
schema := fmt.Sprintf("gl_authority_test_%d", time.Now().UnixNano())
|
||||
if _, err := admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
cleanup, stop := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer stop()
|
||||
if _, err := admin.Exec(cleanup, "DROP SCHEMA "+schema+" CASCADE"); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}()
|
||||
cfg, err := pgxpool.ParseConfig(dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg.ConnConfig.RuntimeParams["search_path"] = schema
|
||||
pool, err := pgxpool.NewWithConfig(ctx, cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer pool.Close()
|
||||
database := &Pool{pool: pool}
|
||||
if err := Migrate(ctx, database); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repository := NewJournalRepository(database)
|
||||
service := applicationledger.NewService(repository, nil)
|
||||
settlementRepo := NewSettlementRepository(database)
|
||||
settlements := applicationsettlement.NewService(settlementRepo)
|
||||
handler := grpcadapter.NewHandler(nil, service, settlements).WithSettlementAdminToken("test-token")
|
||||
authorized := metadata.NewIncomingContext(ctx, metadata.Pairs("x-admin-token", "test-token"))
|
||||
|
||||
account := func(class domain.AccountClass, owner string) domain.AccountReference {
|
||||
return domain.AccountReference{Class: class, OwnerType: "user", OwnerID: owner, AssetID: 7}
|
||||
}
|
||||
command := func(key string, from, to domain.AccountReference, amount string) applicationledger.AppendJournalCommand {
|
||||
credit, err := domain.ParseAmount(amount)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return applicationledger.AppendJournalCommand{
|
||||
SourceService: "internal", SourceTransactionID: key, IdempotencyKey: key, EffectKind: "transfer", EventVersion: 1,
|
||||
ActorID: "operator", OccurredAt: time.Unix(100, 0).UTC(),
|
||||
Entries: []applicationledger.EntryCommand{
|
||||
{LineNumber: 1, Account: from, Amount: credit.Negate().String()},
|
||||
{LineNumber: 2, Account: to, Amount: credit.String()},
|
||||
},
|
||||
}
|
||||
}
|
||||
alice := account(domain.AccountClassUserAvailable, "alice")
|
||||
bob := account(domain.AccountClassUserAvailable, "bob")
|
||||
opening := command("opening", domain.AccountReference{Class: domain.AccountClassTreasury, AssetID: 7}, alice, "100")
|
||||
if _, err := service.AppendJournal(ctx, opening); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
transfer := command("transfer-1", alice, bob, "60")
|
||||
if _, err := service.PostInternalJournal(ctx, transfer); !errors.Is(err, domain.ErrModeRequired) {
|
||||
t.Fatalf("NORMAL admitted internal execution: %v", err)
|
||||
}
|
||||
if _, err := handler.SetOperatingMode(ctx, &ledgerv1.SetOperatingModeRequest{}); status.Code(err) != codes.PermissionDenied {
|
||||
t.Fatalf("unauthorized toggle: %v", err)
|
||||
}
|
||||
if _, _, err := settlements.Enqueue(ctx, "wallet", "t1", "s1", "kuknos", "signed-xdr"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := database.Exec(ctx, "UPDATE kuknos_settlements SET locked_by='in-flight'"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := service.SetOperatingMode(ctx, domain.ModeGLOnly, 0, "operator", "outage", false); !errors.Is(err, domain.ErrSettlementInFlight) {
|
||||
t.Fatalf("in-flight toggle: %v", err)
|
||||
}
|
||||
if _, err := database.Exec(ctx, "UPDATE kuknos_settlements SET locked_by=''"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state, err := handler.SetOperatingMode(authorized, &ledgerv1.SetOperatingModeRequest{Mode: "GL_ONLY", ActorId: "operator", Reason: "outage"})
|
||||
if err != nil || state.GetRevision() != 1 {
|
||||
t.Fatalf("enable GL_ONLY: %+v %v", state, err)
|
||||
}
|
||||
if _, err := service.SetOperatingMode(ctx, domain.ModeNormal, 0, "operator", "stale", true); !errors.Is(err, domain.ErrModeConflict) {
|
||||
t.Fatalf("stale revision accepted: %v", err)
|
||||
}
|
||||
if claimed, err := settlementRepo.ClaimDue(ctx, "worker", 10, time.Now()); err != nil || len(claimed) != 0 {
|
||||
t.Fatalf("settlements claimed in GL_ONLY: %v %v", claimed, err)
|
||||
}
|
||||
if _, _, err := settlements.Enqueue(ctx, "wallet", "t2", "s2", "kuknos", "signed-xdr"); !errors.Is(err, domain.ErrModeRequired) {
|
||||
t.Fatalf("settlement admitted: %v", err)
|
||||
}
|
||||
|
||||
// Two simultaneous spends cannot consume the same opening balance.
|
||||
var wg sync.WaitGroup
|
||||
results := make(chan error, 2)
|
||||
for _, key := range []string{"transfer-1", "transfer-2"} {
|
||||
next := command(key, alice, bob, "60")
|
||||
wg.Add(1)
|
||||
go func(next applicationledger.AppendJournalCommand) {
|
||||
defer wg.Done()
|
||||
_, err := service.PostInternalJournal(ctx, next)
|
||||
results <- err
|
||||
}(next)
|
||||
}
|
||||
wg.Wait()
|
||||
close(results)
|
||||
successes, insufficient := 0, 0
|
||||
for err := range results {
|
||||
if err == nil {
|
||||
successes++
|
||||
} else if errors.Is(err, domain.ErrInsufficientFunds) {
|
||||
insufficient++
|
||||
} else {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if successes != 1 || insufficient != 1 {
|
||||
t.Fatalf("concurrent spends: success=%d insufficient=%d", successes, insufficient)
|
||||
}
|
||||
var successfulKey string
|
||||
if err := database.QueryRow(ctx, "SELECT idempotency_key FROM journals WHERE metadata->>'gl_execution'='GL_ONLY'").Scan(&successfulKey); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
transfer = command(successfulKey, alice, bob, "60")
|
||||
repeated, err := service.PostInternalJournal(ctx, transfer)
|
||||
if err != nil || !repeated.AlreadyExisted {
|
||||
t.Fatalf("idempotent internal retry: %+v %v", repeated, err)
|
||||
}
|
||||
conflicting := command(successfulKey, alice, bob, "1")
|
||||
if _, err := service.PostInternalJournal(ctx, conflicting); !errors.Is(err, domain.ErrIdempotencyConflict) {
|
||||
t.Fatalf("conflicting retry: %v", err)
|
||||
}
|
||||
frozen := account(domain.AccountClassUserFrozen, "alice")
|
||||
if _, err := service.PostInternalJournal(ctx, command("freeze", alice, frozen, "10")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := service.PostInternalJournal(ctx, command("release", frozen, alice, "4")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The legacy ingestion API cannot execute new effects in GL_ONLY.
|
||||
if _, err := service.AppendJournal(ctx, command("legacy-overspend", alice, bob, "100")); !errors.Is(err, domain.ErrModeRequired) {
|
||||
t.Fatalf("legacy bypass: %v", err)
|
||||
}
|
||||
response, err := handler.GetUserBalances(ctx, &ledgerv1.GetUserBalancesRequest{OwnerType: "user", OwnerId: "alice"})
|
||||
if err != nil || len(response.GetBalances()) != 1 {
|
||||
t.Fatalf("user balance: %+v %v", response, err)
|
||||
}
|
||||
balance := response.Balances[0]
|
||||
if balance.Available != "34" || balance.Frozen != "6" || balance.Total != "40" {
|
||||
t.Fatalf("incorrect balance: %+v", balance)
|
||||
}
|
||||
var sum string
|
||||
if err := database.QueryRow(ctx, `SELECT sum(e.amount)::text FROM journal_entries e JOIN ledger_accounts a ON a.id=e.account_id WHERE a.class IN ('USER_AVAILABLE','USER_FROZEN')`).Scan(&sum); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
total, err := domain.ParseAmount(sum)
|
||||
if err != nil || total.String() != "100" {
|
||||
t.Fatalf("conservation failed: %s %v", sum, err)
|
||||
}
|
||||
|
||||
// Exercise the internal RPC mapping with a real commit.
|
||||
posted, err := handler.PostInternalJournal(authorized, &ledgerv1.AppendJournalRequest{
|
||||
SourceService: "internal", SourceTransactionId: "rpc", IdempotencyKey: "rpc", EffectKind: "transfer", EventVersion: 1,
|
||||
ActorId: "operator", OccurredAt: timestamppb.New(time.Unix(100, 0)),
|
||||
Entries: []*ledgerv1.JournalEntry{
|
||||
{LineNumber: 1, Account: &ledgerv1.AccountReference{AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE, OwnerType: "user", OwnerId: "alice", AssetId: 7}, Amount: "-1"},
|
||||
{LineNumber: 2, Account: &ledgerv1.AccountReference{AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE, OwnerType: "user", OwnerId: "bob", AssetId: 7}, Amount: "1"},
|
||||
},
|
||||
})
|
||||
if err != nil || posted.GetJournal().GetJournalId() == "" {
|
||||
t.Fatalf("internal RPC: %+v %v", posted, err)
|
||||
}
|
||||
if _, err := service.SetOperatingMode(ctx, domain.ModeNormal, 1, "operator", "recovery", false); !errors.Is(err, domain.ErrModeConflict) {
|
||||
t.Fatalf("unreconciled recovery: %v", err)
|
||||
}
|
||||
if _, err := service.SetOperatingMode(ctx, domain.ModeNormal, 1, "operator", "recovery", true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repeated, err = service.PostInternalJournal(ctx, transfer)
|
||||
if err != nil || !repeated.AlreadyExisted {
|
||||
t.Fatalf("retry after mode change: %+v %v", repeated, err)
|
||||
}
|
||||
if _, err := service.PostInternalJournal(ctx, command("after-recovery", alice, bob, "1")); !errors.Is(err, domain.ErrModeRequired) {
|
||||
t.Fatalf("internal post after recovery: %v", err)
|
||||
}
|
||||
unknownAsset := int64(99)
|
||||
zero, err := handler.GetUserBalances(ctx, &ledgerv1.GetUserBalancesRequest{OwnerType: "user", OwnerId: "unknown", AssetId: &unknownAsset})
|
||||
if err != nil || len(zero.GetBalances()) != 1 || zero.Balances[0].Total != "0" {
|
||||
t.Fatalf("unknown user balance: %+v %v", zero, err)
|
||||
}
|
||||
// New instances read the same persisted authority state.
|
||||
if persisted, err := NewJournalRepository(database).OperatingMode(ctx); err != nil || persisted.Mode != domain.ModeNormal || persisted.Revision != 2 {
|
||||
t.Fatalf("persisted mode: %+v %v", persisted, err)
|
||||
}
|
||||
if _, err := service.AppendJournal(ctx, command("negative-baseline", alice, bob, "1000")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := service.SetOperatingMode(ctx, domain.ModeGLOnly, 2, "operator", "uncorrected books", false); !errors.Is(err, domain.ErrInsufficientFunds) {
|
||||
t.Fatalf("negative opening balances accepted: %v", err)
|
||||
}
|
||||
var audits int64
|
||||
if err := database.QueryRow(ctx, "SELECT count(*) FROM ledger_mode_changes").Scan(&audits); err != nil || audits != 2 {
|
||||
t.Fatalf("mode audit: %d %v", audits, err)
|
||||
}
|
||||
if _, err := database.Exec(ctx, "DELETE FROM ledger_mode_changes"); err == nil {
|
||||
t.Fatal("mode audit is mutable")
|
||||
}
|
||||
if _, err := database.Exec(ctx, "UPDATE journals SET actor_id='changed'"); err == nil {
|
||||
t.Fatal("journals are mutable")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"gl/domain/ledger"
|
||||
)
|
||||
|
||||
func authorityError(err error) error {
|
||||
var pgErr *pgconn.PgError
|
||||
if errors.As(err, &pgErr) {
|
||||
switch pgErr.Code {
|
||||
case "PGL01":
|
||||
return ledger.ErrModeRequired
|
||||
case "PGL02":
|
||||
return ledger.ErrInsufficientFunds
|
||||
case "40P01", "40001":
|
||||
return ledger.ErrConcurrentPosting
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *JournalRepository) OperatingMode(ctx context.Context) (ledger.OperatingState, error) {
|
||||
return scanOperatingState(r.database.QueryRow(ctx, operatingStateSQL))
|
||||
}
|
||||
|
||||
func scanOperatingState(row Row) (ledger.OperatingState, error) {
|
||||
var state ledger.OperatingState
|
||||
var mode string
|
||||
err := row.Scan(&mode, &state.Revision, &state.ActorID, &state.Reason, &state.ChangedAt)
|
||||
state.Mode = ledger.OperatingMode(mode)
|
||||
return state, err
|
||||
}
|
||||
|
||||
const operatingStateSQL = "SELECT mode, revision, actor_id, reason, changed_at FROM ledger_operating_state WHERE singleton"
|
||||
|
||||
func (r *JournalRepository) SetOperatingMode(ctx context.Context, mode ledger.OperatingMode, revision int64, actor, reason string, reconciled bool) (ledger.OperatingState, error) {
|
||||
tx, err := r.database.Begin(ctx)
|
||||
if err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
current, err := scanOperatingState(tx.QueryRow(ctx, operatingStateSQL+" FOR UPDATE"))
|
||||
if err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
if current.Revision != revision {
|
||||
return ledger.OperatingState{}, ledger.ErrModeConflict
|
||||
}
|
||||
if current.Mode == mode {
|
||||
return current, nil
|
||||
}
|
||||
if mode == ledger.ModeNormal && !reconciled {
|
||||
return ledger.OperatingState{}, fmt.Errorf("%w: reconciliation must be confirmed before returning to NORMAL", ledger.ErrModeConflict)
|
||||
}
|
||||
if mode == ledger.ModeGLOnly {
|
||||
var inFlight bool
|
||||
if err := tx.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM kuknos_settlements WHERE locked_by <> '')").Scan(&inFlight); err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
if inFlight {
|
||||
return ledger.OperatingState{}, ledger.ErrSettlementInFlight
|
||||
}
|
||||
var negativeBalance bool
|
||||
if err := tx.QueryRow(ctx, `SELECT EXISTS (
|
||||
SELECT a.id FROM ledger_accounts a JOIN journal_entries e ON e.account_id=a.id
|
||||
JOIN journals j ON j.id=e.journal_id AND j.sealed_at IS NOT NULL
|
||||
WHERE a.class IN ('USER_AVAILABLE','USER_FROZEN')
|
||||
GROUP BY a.id HAVING sum(e.amount)<0
|
||||
)`).Scan(&negativeBalance); err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
if negativeBalance {
|
||||
return ledger.OperatingState{}, ledger.ErrInsufficientFunds
|
||||
}
|
||||
}
|
||||
state, err := scanOperatingState(tx.QueryRow(ctx, `UPDATE ledger_operating_state
|
||||
SET mode=$1, revision=revision+1, actor_id=$2, reason=$3, changed_at=clock_timestamp()
|
||||
WHERE singleton RETURNING mode, revision, actor_id, reason, changed_at`, string(mode), actor, reason))
|
||||
if err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO ledger_mode_changes
|
||||
(revision, previous_mode, mode, actor_id, reason, reconciliation_confirmed, changed_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7)`, state.Revision, string(current.Mode), string(mode), actor, reason, reconciled, state.ChangedAt); err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return ledger.OperatingState{}, err
|
||||
}
|
||||
slog.WarnContext(ctx, "ledger operating mode changed", "previous_mode", current.Mode, "mode", mode, "revision", state.Revision, "actor_id", actor, "reason", reason)
|
||||
return state, nil
|
||||
}
|
||||
|
||||
// One statement provides a consistent snapshot across available and frozen funds.
|
||||
func (r *JournalRepository) UserBalances(ctx context.Context, ownerType, ownerID string, assetID *int64) ([]ledger.UserAssetBalance, error) {
|
||||
rows, err := r.database.Query(ctx, `SELECT a.asset_id,
|
||||
COALESCE(sum(e.amount) FILTER (WHERE a.class='USER_AVAILABLE'),0)::text,
|
||||
COALESCE(sum(e.amount) FILTER (WHERE a.class='USER_FROZEN'),0)::text
|
||||
FROM ledger_accounts a
|
||||
LEFT JOIN journal_entries e ON e.account_id=a.id
|
||||
LEFT JOIN journals j ON j.id=e.journal_id AND j.sealed_at IS NOT NULL
|
||||
WHERE a.owner_type=$1 AND a.owner_id=$2
|
||||
AND a.class IN ('USER_AVAILABLE','USER_FROZEN')
|
||||
AND ($3::bigint IS NULL OR a.asset_id=$3)
|
||||
AND (e.journal_id IS NULL OR j.id IS NOT NULL)
|
||||
GROUP BY a.asset_id ORDER BY a.asset_id`, ownerType, ownerID, assetID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
result := []ledger.UserAssetBalance{}
|
||||
for rows.Next() {
|
||||
var balance ledger.UserAssetBalance
|
||||
var available, frozen string
|
||||
if err := rows.Scan(&balance.AssetID, &available, &frozen); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
balance.Available, err = ledger.ParseAmount(available)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
balance.Frozen, err = ledger.ParseAmount(frozen)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, balance)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if assetID != nil && len(result) == 0 {
|
||||
result = append(result, ledger.UserAssetBalance{AssetID: *assetID})
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"gl/domain/ledger"
|
||||
|
||||
@@ -47,6 +48,9 @@ func (r *JournalRepository) Append(ctx context.Context, journal ledger.Journal)
|
||||
defer func() {
|
||||
if err != nil {
|
||||
_ = tx.Rollback(ctx)
|
||||
if classified := authorityError(err); classified != nil {
|
||||
err = classified
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -75,13 +79,34 @@ func (r *JournalRepository) Append(ctx context.Context, journal ledger.Journal)
|
||||
}
|
||||
if err != nil {
|
||||
var postgresError *pgconn.PgError
|
||||
if classified := authorityError(err); classified != nil {
|
||||
return AppendResult{}, classified
|
||||
}
|
||||
if errors.As(err, &postgresError) && postgresError.ConstraintName == "journals_one_reversal_idx" {
|
||||
return AppendResult{}, ledger.ErrAlreadyReversed
|
||||
}
|
||||
return AppendResult{}, fmt.Errorf("insert journal: %w", err)
|
||||
}
|
||||
|
||||
for _, entry := range journal.Entries {
|
||||
// Resolve and lock accounts in stable order across every writer.
|
||||
entries := append([]ledger.Entry(nil), journal.Entries...)
|
||||
sort.Slice(entries, func(i, j int) bool {
|
||||
left, right := entries[i].Account, entries[j].Account
|
||||
if left.Class != right.Class {
|
||||
return left.Class < right.Class
|
||||
}
|
||||
if left.OwnerType != right.OwnerType {
|
||||
return left.OwnerType < right.OwnerType
|
||||
}
|
||||
if left.OwnerID != right.OwnerID {
|
||||
return left.OwnerID < right.OwnerID
|
||||
}
|
||||
if left.AssetID != right.AssetID {
|
||||
return left.AssetID < right.AssetID
|
||||
}
|
||||
return entries[i].LineNumber < entries[j].LineNumber
|
||||
})
|
||||
for _, entry := range entries {
|
||||
var accountID int64
|
||||
err = tx.QueryRow(ctx, insertAccountSQL,
|
||||
entry.Account.Class,
|
||||
@@ -114,6 +139,9 @@ func (r *JournalRepository) Append(ctx context.Context, journal ledger.Journal)
|
||||
}
|
||||
|
||||
if _, err = tx.Exec(ctx, "UPDATE journals SET sealed_at = clock_timestamp() WHERE id = $1", journal.ID); err != nil {
|
||||
if classified := authorityError(err); classified != nil {
|
||||
return AppendResult{}, classified
|
||||
}
|
||||
return AppendResult{}, fmt.Errorf("seal journal: %w", err)
|
||||
}
|
||||
if err = tx.Commit(ctx); err != nil {
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
CREATE TABLE ledger_operating_state (
|
||||
singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
|
||||
mode text NOT NULL CHECK (mode IN ('NORMAL','GL_ONLY')),
|
||||
revision bigint NOT NULL CHECK (revision >= 0),
|
||||
actor_id text NOT NULL, reason text NOT NULL,
|
||||
changed_at timestamptz NOT NULL DEFAULT clock_timestamp()
|
||||
);
|
||||
INSERT INTO ledger_operating_state (mode,revision,actor_id,reason) VALUES ('NORMAL',0,'system','initial mode');
|
||||
CREATE TABLE ledger_mode_changes (
|
||||
revision bigint PRIMARY KEY,
|
||||
previous_mode text NOT NULL CHECK (previous_mode IN ('NORMAL','GL_ONLY')),
|
||||
mode text NOT NULL CHECK (mode IN ('NORMAL','GL_ONLY')),
|
||||
actor_id text NOT NULL CHECK (btrim(actor_id) <> ''),
|
||||
reason text NOT NULL CHECK (btrim(reason) <> ''),
|
||||
reconciliation_confirmed boolean NOT NULL, changed_at timestamptz NOT NULL
|
||||
);
|
||||
CREATE TRIGGER mode_changes_reject_mutation BEFORE UPDATE OR DELETE ON ledger_mode_changes
|
||||
FOR EACH ROW EXECUTE FUNCTION reject_ledger_mutation();
|
||||
CREATE TRIGGER operating_state_reject_delete BEFORE DELETE ON ledger_operating_state
|
||||
FOR EACH ROW EXECUTE FUNCTION reject_ledger_mutation();
|
||||
|
||||
-- A shared lock is held through each posting; toggles drain writers.
|
||||
CREATE FUNCTION guard_journal_authority() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
DECLARE current_mode text;
|
||||
BEGIN
|
||||
SELECT mode INTO STRICT current_mode FROM ledger_operating_state WHERE singleton FOR SHARE;
|
||||
IF NEW.metadata->>'gl_execution'='GL_ONLY' AND current_mode<>'GL_ONLY'
|
||||
AND NOT EXISTS (SELECT 1 FROM journals WHERE idempotency_key=NEW.idempotency_key) THEN
|
||||
RAISE EXCEPTION 'GL_ONLY mode is required' USING ERRCODE='PGL01';
|
||||
END IF;
|
||||
IF current_mode='GL_ONLY' AND COALESCE(NEW.metadata->>'gl_execution','')<>'GL_ONLY'
|
||||
AND NOT EXISTS (SELECT 1 FROM journals WHERE idempotency_key=NEW.idempotency_key) THEN
|
||||
RAISE EXCEPTION 'new GL_ONLY effects must use internal execution' USING ERRCODE='PGL01';
|
||||
END IF;
|
||||
RETURN NEW;
|
||||
END; $$;
|
||||
CREATE TRIGGER journals_guard_authority BEFORE INSERT ON journals
|
||||
FOR EACH ROW EXECUTE FUNCTION guard_journal_authority();
|
||||
|
||||
-- Both credits and debits lock accounts, including normal recorders.
|
||||
CREATE OR REPLACE FUNCTION guard_entry_insert() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
IF (SELECT sealed_at IS NOT NULL FROM journals WHERE id=NEW.journal_id) THEN
|
||||
RAISE EXCEPTION 'cannot append to a sealed journal' USING ERRCODE='55000';
|
||||
END IF;
|
||||
PERFORM id FROM ledger_accounts WHERE id=NEW.account_id
|
||||
AND class IN ('USER_AVAILABLE','USER_FROZEN') FOR UPDATE;
|
||||
RETURN NEW;
|
||||
END; $$;
|
||||
|
||||
CREATE OR REPLACE FUNCTION guard_journal_seal() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
DECLARE current_mode text;
|
||||
BEGIN
|
||||
IF OLD.sealed_at IS NOT NULL OR NEW.sealed_at IS NULL
|
||||
OR (to_jsonb(NEW)-'sealed_at') IS DISTINCT FROM (to_jsonb(OLD)-'sealed_at') THEN
|
||||
RAISE EXCEPTION 'journals are immutable except for initial sealing' USING ERRCODE='55000';
|
||||
END IF;
|
||||
IF (SELECT count(*) FROM journal_entries WHERE journal_id=NEW.id)<2 THEN
|
||||
RAISE EXCEPTION 'journal requires at least two entries' USING ERRCODE='23514';
|
||||
END IF;
|
||||
IF EXISTS (SELECT asset_id FROM journal_entries WHERE journal_id=NEW.id GROUP BY asset_id HAVING sum(amount)<>0) THEN
|
||||
RAISE EXCEPTION 'journal is not balanced per asset' USING ERRCODE='23514';
|
||||
END IF;
|
||||
SELECT mode INTO STRICT current_mode FROM ledger_operating_state WHERE singleton;
|
||||
IF current_mode='GL_ONLY' THEN
|
||||
IF NEW.blockchain_network<>'' OR NEW.blockchain_transaction_hash<>'' OR NEW.blockchain_ledger_sequence<>''
|
||||
OR EXISTS (SELECT 1 FROM journal_entries e JOIN ledger_accounts a ON a.id=e.account_id
|
||||
WHERE e.journal_id=NEW.id AND a.class='EXTERNAL_BLOCKCHAIN') THEN
|
||||
RAISE EXCEPTION 'external blockchain movements are unavailable in GL_ONLY' USING ERRCODE='PGL01';
|
||||
END IF;
|
||||
IF EXISTS (
|
||||
SELECT a.id FROM ledger_accounts a JOIN journal_entries e ON e.account_id=a.id JOIN journals j ON j.id=e.journal_id
|
||||
WHERE a.class IN ('USER_AVAILABLE','USER_FROZEN')
|
||||
AND a.id IN (SELECT account_id FROM journal_entries WHERE journal_id=NEW.id)
|
||||
AND (j.sealed_at IS NOT NULL OR j.id=NEW.id)
|
||||
GROUP BY a.id HAVING sum(e.amount)<0
|
||||
) THEN
|
||||
RAISE EXCEPTION 'insufficient ledger funds' USING ERRCODE='PGL02';
|
||||
END IF;
|
||||
END IF;
|
||||
RETURN NEW;
|
||||
END; $$;
|
||||
|
||||
CREATE FUNCTION guard_settlement_authority() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
DECLARE current_mode text;
|
||||
BEGIN
|
||||
SELECT mode INTO STRICT current_mode FROM ledger_operating_state WHERE singleton FOR SHARE;
|
||||
IF current_mode='GL_ONLY'
|
||||
AND NOT EXISTS (SELECT 1 FROM kuknos_settlements WHERE idempotency_key=NEW.idempotency_key) THEN
|
||||
RAISE EXCEPTION 'blockchain settlement is disabled in GL_ONLY' USING ERRCODE='PGL01';
|
||||
END IF;
|
||||
RETURN NEW;
|
||||
END; $$;
|
||||
CREATE TRIGGER settlements_guard_authority BEFORE INSERT ON kuknos_settlements
|
||||
FOR EACH ROW EXECUTE FUNCTION guard_settlement_authority();
|
||||
@@ -23,6 +23,14 @@ func NewSettlementRepository(database Database) *SettlementRepository {
|
||||
return &SettlementRepository{database: database}
|
||||
}
|
||||
|
||||
// Existing normal-mode Wallet clients require SettlementReady; a GL-only
|
||||
// cutover must not let them continue admitting blockchain-dependent effects.
|
||||
func (r *SettlementRepository) SettlementAllowed(ctx context.Context) (bool, error) {
|
||||
var mode string
|
||||
err := r.database.QueryRow(ctx, "SELECT mode FROM ledger_operating_state WHERE singleton").Scan(&mode)
|
||||
return mode == string(ledger.ModeNormal), err
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) Stats(ctx context.Context) (ledger.SettlementStats, error) {
|
||||
var stats ledger.SettlementStats
|
||||
err := r.database.QueryRow(ctx, `SELECT
|
||||
@@ -60,6 +68,9 @@ func (r *SettlementRepository) Enqueue(ctx context.Context, s ledger.Settlement)
|
||||
existing, err := r.Get(ctx, inserted, "")
|
||||
return existing, true, err
|
||||
}
|
||||
if classified := authorityError(err); classified != nil {
|
||||
return ledger.Settlement{}, false, classified
|
||||
}
|
||||
return ledger.Settlement{}, false, fmt.Errorf("enqueue settlement: %w", err)
|
||||
}
|
||||
|
||||
@@ -78,6 +89,13 @@ func (r *SettlementRepository) ClaimDue(ctx context.Context, workerID string, li
|
||||
return nil, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var mode string
|
||||
if err := tx.QueryRow(ctx, "SELECT mode FROM ledger_operating_state WHERE singleton FOR SHARE").Scan(&mode); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if mode == string(ledger.ModeGLOnly) {
|
||||
return nil, nil
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `UPDATE kuknos_settlements SET locked_at=NULL, locked_by='', updated_at=clock_timestamp() WHERE locked_at < clock_timestamp() - interval '10 minutes'`); err != nil {
|
||||
return nil, fmt.Errorf("release stale settlement locks: %w", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user