feat: add durable Kuknos settlement coordination
This commit is contained in:
Navid
@@ -3,6 +3,8 @@ package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/knadh/koanf/parsers/toml"
|
||||
@@ -11,21 +13,22 @@ import (
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Environment string `koanf:"environment"`
|
||||
GRPC GRPCConfig `koanf:"grpc"`
|
||||
Database DatabaseConfig `koanf:"database"`
|
||||
Environment string `koanf:"environment"`
|
||||
GRPC GRPCConfig `koanf:"grpc"`
|
||||
Database DatabaseConfig `koanf:"db"`
|
||||
Settlement SettlementConfig `koanf:"settlement"`
|
||||
}
|
||||
|
||||
type DashboardConfig struct {
|
||||
Environment string `koanf:"environment"`
|
||||
HTTP HTTPConfig `koanf:"http"`
|
||||
Database DatabaseConfig `koanf:"database"`
|
||||
Database DatabaseConfig `koanf:"db"`
|
||||
}
|
||||
|
||||
type GRPCConfig struct {
|
||||
Host string `koanf:"host"`
|
||||
Port int `koanf:"port"`
|
||||
ShutdownTimeout time.Duration `koanf:"shutdown-timeout"`
|
||||
ShutdownTimeout time.Duration `koanf:"timeout"`
|
||||
}
|
||||
|
||||
type HTTPConfig struct {
|
||||
@@ -36,12 +39,23 @@ type HTTPConfig struct {
|
||||
}
|
||||
|
||||
type DatabaseConfig struct {
|
||||
Host string `koanf:"host"`
|
||||
Port int `koanf:"port"`
|
||||
Name string `koanf:"name"`
|
||||
User string `koanf:"user"`
|
||||
Password string `koanf:"password"`
|
||||
SSLMode string `koanf:"ssl-mode"`
|
||||
Host string `koanf:"host"`
|
||||
Port int `koanf:"port"`
|
||||
Name string `koanf:"name"`
|
||||
User string `koanf:"user"`
|
||||
Password string `koanf:"password"`
|
||||
GormLogLevel int `koanf:"gorm-log-level"`
|
||||
SSLMode string `koanf:"-"`
|
||||
}
|
||||
|
||||
type SettlementConfig struct {
|
||||
Enabled bool `koanf:"enabled"`
|
||||
Endpoint string `koanf:"endpoint"`
|
||||
WorkerID string `koanf:"worker-id"`
|
||||
PollInterval time.Duration `koanf:"poll-interval"`
|
||||
BatchSize int `koanf:"batch-size"`
|
||||
MaxAttempts int `koanf:"max-attempts"`
|
||||
AdminToken string `koanf:"admin-token"`
|
||||
}
|
||||
|
||||
func Load(path string) (*Config, error) {
|
||||
@@ -52,12 +66,14 @@ func Load(path string) (*Config, error) {
|
||||
Port: 8600,
|
||||
ShutdownTimeout: 10 * time.Second,
|
||||
},
|
||||
Database: defaultDatabaseConfig(),
|
||||
Database: defaultDatabaseConfig(),
|
||||
Settlement: SettlementConfig{WorkerID: "gl-settlement-1", PollInterval: 5 * time.Second, BatchSize: 20, MaxAttempts: 8},
|
||||
}
|
||||
|
||||
if err := load(path, cfg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
applyEnvironment(&cfg.Database, &cfg.Settlement)
|
||||
if err := cfg.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -78,12 +94,28 @@ func LoadDashboard(path string) (*DashboardConfig, error) {
|
||||
if err := load(path, cfg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
applyEnvironment(&cfg.Database, nil)
|
||||
if err := cfg.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func applyEnvironment(database *DatabaseConfig, settlement *SettlementConfig) {
|
||||
if value, ok := os.LookupEnv("DARANO_GL_DB_PASSWORD"); ok {
|
||||
database.Password = value
|
||||
}
|
||||
if settlement == nil {
|
||||
return
|
||||
}
|
||||
if value, ok := os.LookupEnv("DARANO_GL_SETTLEMENT_ENDPOINT"); ok {
|
||||
settlement.Endpoint = strings.TrimRight(value, "/")
|
||||
}
|
||||
if value, ok := os.LookupEnv("DARANO_GL_ADMIN_TOKEN"); ok {
|
||||
settlement.AdminToken = value
|
||||
}
|
||||
}
|
||||
|
||||
func load(path string, target any) error {
|
||||
k := koanf.New(".")
|
||||
if err := k.Load(file.Provider(path), toml.Parser()); err != nil {
|
||||
@@ -115,6 +147,9 @@ func (c *Config) Validate() error {
|
||||
if c.GRPC.ShutdownTimeout <= 0 {
|
||||
return fmt.Errorf("grpc shutdown timeout must be positive")
|
||||
}
|
||||
if c.Environment == "production" && c.Settlement.AdminToken == "" {
|
||||
return fmt.Errorf("settlement admin token is required in production")
|
||||
}
|
||||
return validateDatabase(c.Database)
|
||||
}
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ import (
|
||||
func TestLoadUsesDefaultsAndOverrides(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "gl.toml")
|
||||
contents := []byte("[grpc]\nport = 0\nshutdown-timeout = \"3s\"\n")
|
||||
contents := []byte("[grpc]\nport = 0\ntimeout = \"3s\"\n")
|
||||
if err := os.WriteFile(path, contents, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -30,7 +30,7 @@ func TestLoadUsesDefaultsAndOverrides(t *testing.T) {
|
||||
func TestLoadDashboardUsesHTTPDefaultsAndOverrides(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "dashboard.toml")
|
||||
contents := []byte("[http]\nport = 0\nshutdown-timeout = \"3s\"\n")
|
||||
contents := []byte("[http]\nport = 0\nshutdown-timeout = \"3s\"\n[db]\nname = \"dashboard_db\"\n")
|
||||
if err := os.WriteFile(path, contents, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -42,7 +42,7 @@ func TestLoadDashboardUsesHTTPDefaultsAndOverrides(t *testing.T) {
|
||||
if cfg.HTTP.Port != 0 || cfg.HTTP.ShutdownTimeout != 3*time.Second || cfg.HTTP.ReadHeaderTimeout != 5*time.Second {
|
||||
t.Fatalf("unexpected http config: %+v", cfg.HTTP)
|
||||
}
|
||||
if cfg.Database.Name != "gl_db" || cfg.Database.Port != 5432 {
|
||||
if cfg.Database.Name != "dashboard_db" || cfg.Database.Port != 5432 {
|
||||
t.Fatalf("unexpected database defaults: %+v", cfg.Database)
|
||||
}
|
||||
}
|
||||
@@ -50,7 +50,7 @@ func TestLoadDashboardUsesHTTPDefaultsAndOverrides(t *testing.T) {
|
||||
func TestLoadRejectsInvalidConfiguration(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "gl.toml")
|
||||
if err := os.WriteFile(path, []byte("[grpc]\nshutdown-timeout = \"0s\"\n"), 0o600); err != nil {
|
||||
if err := os.WriteFile(path, []byte("[grpc]\ntimeout = \"0s\"\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -64,3 +64,46 @@ func TestLoadReturnsMissingFileError(t *testing.T) {
|
||||
t.Fatal("expected missing file error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadAppliesSecretEnvironmentOverrides(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "gl.toml")
|
||||
contents := []byte("environment = \"production\"\n[grpc]\ntimeout = \"3s\"\n[settlement]\nenabled = true\nendpoint = \"https://invalid.example\"\n")
|
||||
if err := os.WriteFile(path, contents, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Setenv("DARANO_GL_DB_PASSWORD", "runtime-db-secret")
|
||||
t.Setenv("DARANO_GL_ADMIN_TOKEN", "runtime-admin-secret")
|
||||
t.Setenv("DARANO_GL_SETTLEMENT_ENDPOINT", "https://horizon.example/")
|
||||
|
||||
cfg, err := Load(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.Database.Password != "runtime-db-secret" {
|
||||
t.Fatal("database password was not loaded from the environment")
|
||||
}
|
||||
if cfg.Settlement.AdminToken != "runtime-admin-secret" {
|
||||
t.Fatal("admin token was not loaded from the environment")
|
||||
}
|
||||
if cfg.Settlement.Endpoint != "https://horizon.example" {
|
||||
t.Fatalf("unexpected settlement endpoint: %q", cfg.Settlement.Endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadDashboardAppliesDatabasePasswordEnvironmentOverride(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "dashboard.toml")
|
||||
if err := os.WriteFile(path, []byte("[http]\nport = 0\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Setenv("DARANO_GL_DB_PASSWORD", "runtime-db-secret")
|
||||
|
||||
cfg, err := LoadDashboard(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.Database.Password != "runtime-db-secret" {
|
||||
t.Fatal("database password was not loaded from the environment")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package kuknos
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
||||
"gl/domain/ledger"
|
||||
)
|
||||
|
||||
// Submitter sends a pre-signed Stellar/Kuknos transaction to Horizon. Signing
|
||||
// remains outside GL; GL owns durable admission, retry, and reconciliation.
|
||||
type Submitter struct {
|
||||
Endpoint string
|
||||
Client *http.Client
|
||||
}
|
||||
|
||||
func (s Submitter) Submit(ctx context.Context, record ledger.Settlement) (string, error) {
|
||||
if s.Endpoint == "" || record.SignedTransactionXDR == "" {
|
||||
return "", fmt.Errorf("kuknos submission endpoint and signed xdr are required")
|
||||
}
|
||||
form := url.Values{"tx": {record.SignedTransactionXDR}}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.Endpoint, "/")+"/transactions", strings.NewReader(form.Encode()))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
|
||||
client := s.Client
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var body struct {
|
||||
Hash string `json:"hash"`
|
||||
Extras struct {
|
||||
ResultCodes struct {
|
||||
Transaction string `json:"transaction"`
|
||||
} `json:"result_codes"`
|
||||
} `json:"extras"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
|
||||
return "", fmt.Errorf("decode kuknos response: %w", err)
|
||||
}
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 || body.Hash == "" {
|
||||
return "", fmt.Errorf("kuknos rejected transaction: status=%d code=%s", resp.StatusCode, body.Extras.ResultCodes.Transaction)
|
||||
}
|
||||
return body.Hash, nil
|
||||
}
|
||||
@@ -98,3 +98,7 @@ type txAdapter struct {
|
||||
func (t txAdapter) QueryRow(ctx context.Context, sql string, args ...any) Row {
|
||||
return t.Tx.QueryRow(ctx, sql, args...)
|
||||
}
|
||||
|
||||
func (t txAdapter) Query(ctx context.Context, sql string, args ...any) (Rows, error) {
|
||||
return t.Tx.Query(ctx, sql, args...)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
CREATE TABLE kuknos_settlements (
|
||||
id uuid PRIMARY KEY,
|
||||
source_service text NOT NULL CHECK (source_service <> ''),
|
||||
source_transaction_id text NOT NULL CHECK (source_transaction_id <> ''),
|
||||
idempotency_key text NOT NULL UNIQUE CHECK (idempotency_key <> ''),
|
||||
status text NOT NULL CHECK (status IN ('PENDING','SUBMITTED','CONFIRMED','RETRYABLE','MANUAL_REVIEW')),
|
||||
network text NOT NULL DEFAULT 'kuknos',
|
||||
signed_transaction_xdr text NOT NULL DEFAULT '',
|
||||
transaction_hash text NOT NULL DEFAULT '',
|
||||
attempts integer NOT NULL DEFAULT 0 CHECK (attempts >= 0),
|
||||
available_at timestamptz NOT NULL DEFAULT clock_timestamp(),
|
||||
locked_at timestamptz,
|
||||
locked_by text NOT NULL DEFAULT '',
|
||||
last_error text NOT NULL DEFAULT '',
|
||||
submitted_at timestamptz,
|
||||
confirmed_at timestamptz,
|
||||
manual_review_at timestamptz,
|
||||
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
|
||||
updated_at timestamptz NOT NULL DEFAULT clock_timestamp()
|
||||
);
|
||||
|
||||
CREATE INDEX kuknos_settlements_due_idx ON kuknos_settlements (status, available_at);
|
||||
@@ -0,0 +1,13 @@
|
||||
CREATE TABLE kuknos_settlement_operator_audits (
|
||||
id bigserial PRIMARY KEY,
|
||||
settlement_id uuid NOT NULL REFERENCES kuknos_settlements(id),
|
||||
actor_id text NOT NULL CHECK (actor_id <> ''),
|
||||
action text NOT NULL CHECK (action IN ('RETRY')),
|
||||
reason text NOT NULL CHECK (reason <> ''),
|
||||
previous_status text NOT NULL,
|
||||
resulting_status text NOT NULL,
|
||||
created_at timestamptz NOT NULL DEFAULT clock_timestamp()
|
||||
);
|
||||
|
||||
CREATE INDEX kuknos_settlement_operator_audits_settlement_idx
|
||||
ON kuknos_settlement_operator_audits (settlement_id, created_at DESC);
|
||||
@@ -0,0 +1,13 @@
|
||||
CREATE TABLE kuknos_reconciliation_issues (
|
||||
settlement_id uuid PRIMARY KEY REFERENCES kuknos_settlements(id),
|
||||
issue_type text NOT NULL CHECK (issue_type IN ('MISSING_LEDGER_CONFIRMATION','TRANSACTION_HASH_MISMATCH')),
|
||||
expected_hash text NOT NULL DEFAULT '',
|
||||
actual_hash text NOT NULL DEFAULT '',
|
||||
status text NOT NULL DEFAULT 'OPEN' CHECK (status IN ('OPEN','RESOLVED')),
|
||||
first_detected_at timestamptz NOT NULL DEFAULT clock_timestamp(),
|
||||
last_detected_at timestamptz NOT NULL DEFAULT clock_timestamp(),
|
||||
resolved_at timestamptz
|
||||
);
|
||||
|
||||
CREATE INDEX kuknos_reconciliation_issues_open_idx
|
||||
ON kuknos_reconciliation_issues (status, last_detected_at DESC);
|
||||
@@ -0,0 +1,204 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"gl/domain/ledger"
|
||||
)
|
||||
|
||||
type SettlementRepository struct{ database Database }
|
||||
|
||||
type ReconciliationStats struct {
|
||||
Open, Detected, Resolved int64
|
||||
}
|
||||
|
||||
type txQueryer interface {
|
||||
Query(context.Context, string, ...any) (Rows, error)
|
||||
}
|
||||
|
||||
func NewSettlementRepository(database Database) *SettlementRepository {
|
||||
return &SettlementRepository{database: database}
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) Stats(ctx context.Context) (ledger.SettlementStats, error) {
|
||||
var stats ledger.SettlementStats
|
||||
err := r.database.QueryRow(ctx, `SELECT
|
||||
COUNT(*) FILTER (WHERE status='PENDING'),
|
||||
COUNT(*) FILTER (WHERE status='RETRYABLE'),
|
||||
COUNT(*) FILTER (WHERE status='MANUAL_REVIEW'),
|
||||
COALESCE(EXTRACT(EPOCH FROM (clock_timestamp() - MIN(created_at) FILTER (WHERE status IN ('PENDING','RETRYABLE'))))::bigint, 0)
|
||||
FROM kuknos_settlements`).Scan(&stats.Pending, &stats.Retryable, &stats.ManualReview, &stats.OldestPendingSeconds)
|
||||
if err != nil {
|
||||
return ledger.SettlementStats{}, fmt.Errorf("settlement stats: %w", err)
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) Enqueue(ctx context.Context, s ledger.Settlement) (ledger.Settlement, bool, error) {
|
||||
if err := s.Validate(); err != nil {
|
||||
return ledger.Settlement{}, false, err
|
||||
}
|
||||
var inserted string
|
||||
err := r.database.QueryRow(ctx, `INSERT INTO kuknos_settlements
|
||||
(id, source_service, source_transaction_id, idempotency_key, status, network, signed_transaction_xdr, available_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT (idempotency_key) DO NOTHING RETURNING id`,
|
||||
s.ID, s.SourceService, s.SourceTxID, s.IdempotencyKey, s.Status, s.Network, s.SignedTransactionXDR, s.AvailableAt).Scan(&inserted)
|
||||
if err == nil {
|
||||
return s, false, nil
|
||||
}
|
||||
if err == pgx.ErrNoRows {
|
||||
var existingXDR string
|
||||
if err := r.database.QueryRow(ctx, `SELECT signed_transaction_xdr, id FROM kuknos_settlements WHERE idempotency_key=$1`, s.IdempotencyKey).Scan(&existingXDR, &inserted); err != nil {
|
||||
return ledger.Settlement{}, false, fmt.Errorf("resolve settlement idempotency: %w", err)
|
||||
}
|
||||
if existingXDR != s.SignedTransactionXDR {
|
||||
return ledger.Settlement{}, false, ledger.ErrIdempotencyConflict
|
||||
}
|
||||
existing, err := r.Get(ctx, inserted, "")
|
||||
return existing, true, err
|
||||
}
|
||||
return ledger.Settlement{}, false, fmt.Errorf("enqueue settlement: %w", err)
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) Get(ctx context.Context, id, idempotencyKey string) (ledger.Settlement, error) {
|
||||
var s ledger.Settlement
|
||||
err := r.database.QueryRow(ctx, `SELECT id, source_service, source_transaction_id, idempotency_key, status, network, transaction_hash, attempts, available_at, last_error, signed_transaction_xdr FROM kuknos_settlements WHERE id=NULLIF($1,'')::uuid OR idempotency_key=NULLIF($2,'')`, id, idempotencyKey).Scan(&s.ID, &s.SourceService, &s.SourceTxID, &s.IdempotencyKey, &s.Status, &s.Network, &s.TransactionHash, &s.Attempts, &s.AvailableAt, &s.LastError, &s.SignedTransactionXDR)
|
||||
if err != nil {
|
||||
return ledger.Settlement{}, fmt.Errorf("get settlement: %w", err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) ClaimDue(ctx context.Context, workerID string, limit int, now time.Time) ([]ledger.Settlement, error) {
|
||||
tx, err := r.database.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
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)
|
||||
}
|
||||
queryer, ok := tx.(txQueryer)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("settlement transaction does not support row queries")
|
||||
}
|
||||
rows, err := queryer.Query(ctx, `WITH due AS (
|
||||
SELECT id FROM kuknos_settlements
|
||||
WHERE status IN ('PENDING','RETRYABLE') AND available_at <= $1 AND locked_by=''
|
||||
ORDER BY available_at, created_at FOR UPDATE SKIP LOCKED LIMIT $2
|
||||
)
|
||||
UPDATE kuknos_settlements s SET locked_at=clock_timestamp(), locked_by=$3, updated_at=clock_timestamp()
|
||||
FROM due WHERE s.id=due.id
|
||||
RETURNING s.id, s.source_service, s.source_transaction_id, s.idempotency_key, s.status, s.network, s.transaction_hash, s.attempts, s.available_at, s.last_error, s.signed_transaction_xdr`, now, limit, workerID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var result []ledger.Settlement
|
||||
for rows.Next() {
|
||||
var s ledger.Settlement
|
||||
if err := rows.Scan(&s.ID, &s.SourceService, &s.SourceTxID, &s.IdempotencyKey, &s.Status, &s.Network, &s.TransactionHash, &s.Attempts, &s.AvailableAt, &s.LastError, &s.SignedTransactionXDR); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, s)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) Save(ctx context.Context, s ledger.Settlement, workerID string) error {
|
||||
_, err := r.database.Exec(ctx, `UPDATE kuknos_settlements SET status=$2, transaction_hash=$3, attempts=$4, available_at=$5, last_error=$6, locked_at=NULL, locked_by='', submitted_at=CASE WHEN $2 IN ('SUBMITTED','CONFIRMED') THEN COALESCE(submitted_at, clock_timestamp()) ELSE submitted_at END, confirmed_at=CASE WHEN $2='CONFIRMED' THEN COALESCE(confirmed_at, clock_timestamp()) ELSE confirmed_at END, manual_review_at=CASE WHEN $2='MANUAL_REVIEW' THEN clock_timestamp() ELSE manual_review_at END, updated_at=clock_timestamp() WHERE id=$1 AND locked_by=$7`, s.ID, s.Status, s.TransactionHash, s.Attempts, s.AvailableAt, s.LastError, workerID)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *SettlementRepository) RetryByOperator(ctx context.Context, id, actorID, reason string, now time.Time) (ledger.Settlement, error) {
|
||||
tx, err := r.database.Begin(ctx)
|
||||
if err != nil {
|
||||
return ledger.Settlement{}, fmt.Errorf("begin operator retry: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var previous ledger.SettlementStatus
|
||||
if err := tx.QueryRow(ctx, `SELECT status FROM kuknos_settlements WHERE id=$1 FOR UPDATE`, id).Scan(&previous); err != nil {
|
||||
return ledger.Settlement{}, fmt.Errorf("lock settlement for operator retry: %w", err)
|
||||
}
|
||||
if previous == ledger.SettlementConfirmed {
|
||||
return ledger.Settlement{}, fmt.Errorf("confirmed settlement cannot be retried")
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `UPDATE kuknos_settlements SET status='RETRYABLE', available_at=$2, locked_at=NULL, locked_by='', last_error='', manual_review_at=NULL, updated_at=clock_timestamp() WHERE id=$1`, id, now); err != nil {
|
||||
return ledger.Settlement{}, fmt.Errorf("retry settlement: %w", err)
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO kuknos_settlement_operator_audits (settlement_id, actor_id, action, reason, previous_status, resulting_status) VALUES ($1,$2,'RETRY',$3,$4,'RETRYABLE')`, id, actorID, reason, previous); err != nil {
|
||||
return ledger.Settlement{}, fmt.Errorf("audit operator retry: %w", err)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return ledger.Settlement{}, fmt.Errorf("commit operator retry: %w", err)
|
||||
}
|
||||
return r.Get(ctx, id, "")
|
||||
}
|
||||
|
||||
// ReconcileConfirmationEvidence durably records confirmed Kuknos settlements
|
||||
// whose latest GL lifecycle evidence is missing or has a different hash.
|
||||
func (r *SettlementRepository) ReconcileConfirmationEvidence(ctx context.Context, grace time.Duration) (ReconciliationStats, error) {
|
||||
if grace <= 0 {
|
||||
grace = 10 * time.Minute
|
||||
}
|
||||
tx, err := r.database.Begin(ctx)
|
||||
if err != nil {
|
||||
return ReconciliationStats{}, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var detected int64
|
||||
err = tx.QueryRow(ctx, `WITH inconsistent AS (
|
||||
SELECT s.id,
|
||||
CASE WHEN e.id IS NULL OR e.blockchain_transaction_hash='' THEN 'MISSING_LEDGER_CONFIRMATION' ELSE 'TRANSACTION_HASH_MISMATCH' END issue_type,
|
||||
s.transaction_hash expected_hash, COALESCE(e.blockchain_transaction_hash,'') actual_hash
|
||||
FROM kuknos_settlements s
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT id, blockchain_transaction_hash FROM transaction_events
|
||||
WHERE source_service=s.source_service AND source_transaction_id=s.source_transaction_id
|
||||
ORDER BY event_version DESC, recorded_at DESC LIMIT 1
|
||||
) e ON true
|
||||
WHERE s.status='CONFIRMED' AND s.confirmed_at < clock_timestamp() - $1::interval
|
||||
AND (e.id IS NULL OR e.blockchain_transaction_hash='' OR e.blockchain_transaction_hash<>s.transaction_hash)
|
||||
), upserted AS (
|
||||
INSERT INTO kuknos_reconciliation_issues (settlement_id, issue_type, expected_hash, actual_hash)
|
||||
SELECT id, issue_type, expected_hash, actual_hash FROM inconsistent
|
||||
ON CONFLICT (settlement_id) DO UPDATE SET issue_type=EXCLUDED.issue_type,
|
||||
expected_hash=EXCLUDED.expected_hash, actual_hash=EXCLUDED.actual_hash,
|
||||
status='OPEN', last_detected_at=clock_timestamp(), resolved_at=NULL
|
||||
WHERE kuknos_reconciliation_issues.status<>'OPEN'
|
||||
OR kuknos_reconciliation_issues.issue_type<>EXCLUDED.issue_type
|
||||
OR kuknos_reconciliation_issues.expected_hash<>EXCLUDED.expected_hash
|
||||
OR kuknos_reconciliation_issues.actual_hash<>EXCLUDED.actual_hash
|
||||
RETURNING 1
|
||||
) SELECT count(*) FROM upserted`, fmt.Sprintf("%f seconds", grace.Seconds())).Scan(&detected)
|
||||
if err != nil {
|
||||
return ReconciliationStats{}, fmt.Errorf("detect settlement reconciliation issues: %w", err)
|
||||
}
|
||||
result, err := tx.Exec(ctx, `UPDATE kuknos_reconciliation_issues i SET status='RESOLVED', resolved_at=clock_timestamp()
|
||||
WHERE status='OPEN' AND EXISTS (
|
||||
SELECT 1 FROM kuknos_settlements s JOIN transaction_events e
|
||||
ON e.source_service=s.source_service AND e.source_transaction_id=s.source_transaction_id
|
||||
WHERE s.id=i.settlement_id AND s.status='CONFIRMED'
|
||||
AND e.blockchain_transaction_hash=s.transaction_hash AND e.blockchain_transaction_hash<>''
|
||||
)`)
|
||||
if err != nil {
|
||||
return ReconciliationStats{}, fmt.Errorf("resolve settlement reconciliation issues: %w", err)
|
||||
}
|
||||
var open int64
|
||||
if err := tx.QueryRow(ctx, `SELECT count(*) FROM kuknos_reconciliation_issues WHERE status='OPEN'`).Scan(&open); err != nil {
|
||||
return ReconciliationStats{}, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return ReconciliationStats{}, err
|
||||
}
|
||||
return ReconciliationStats{Open: open, Detected: detected, Resolved: result.RowsAffected()}, nil
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestReconcileConfirmationEvidenceCommitsDurableIssueStats(t *testing.T) {
|
||||
tx := &fakeTx{rows: []Row{
|
||||
fakeRow{values: []any{int64(1)}},
|
||||
fakeRow{values: []any{int64(2)}},
|
||||
}}
|
||||
repository := NewSettlementRepository(&fakeDatabase{tx: tx})
|
||||
|
||||
stats, err := repository.ReconcileConfirmationEvidence(context.Background(), 10*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stats.Detected != 1 || stats.Resolved != 1 || stats.Open != 2 {
|
||||
t.Fatalf("unexpected reconciliation stats: %+v", stats)
|
||||
}
|
||||
if !tx.committed || tx.execCount != 1 {
|
||||
t.Fatalf("unexpected transaction state: %+v", tx)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcileConfirmationEvidenceRollsBackResolutionFailure(t *testing.T) {
|
||||
tx := &fakeTx{
|
||||
rows: []Row{fakeRow{values: []any{int64(1)}}},
|
||||
execErrAt: 1,
|
||||
}
|
||||
repository := NewSettlementRepository(&fakeDatabase{tx: tx})
|
||||
|
||||
if _, err := repository.ReconcileConfirmationEvidence(context.Background(), time.Minute); err == nil {
|
||||
t.Fatal("expected resolution error")
|
||||
}
|
||||
if !tx.rolledBack || tx.committed {
|
||||
t.Fatalf("failed reconciliation was not rolled back: %+v", tx)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user