b5coordinator

package
v1.5.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 17, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package b5coordinator implements the feature-off B5 planned transaction coordinator. It deliberately has no MCP, HTTP, feature-flag, credential or raw database dependency. The only executable object it can receive is the narrow, transaction-bound Capability below.

Index

Constants

View Source
const (
	PlanSchemaID      = "agentsql.b5.plan.v4"
	PlanSchemaVersion = uint16(4)
	BeginAuthSchemaID = "agentsql.b5.begin-auth.v3"
	RequiredBinderABI = b5dml.BinderABI
)

Variables

View Source
var (
	ErrInvalidPlan       = errors.New("b5coordinator: invalid transaction plan")
	ErrPlanDenied        = errors.New("b5coordinator: plan denied")
	ErrApprovalRequired  = errors.New("b5coordinator: approval required")
	ErrApprovalInvalid   = errors.New("b5coordinator: approval invalid")
	ErrWrongOwner        = errors.New("b5coordinator: wrong owner or epoch")
	ErrBusy              = errors.New("b5coordinator: operation already active")
	ErrWrongOrdinal      = errors.New("b5coordinator: operation is not next in plan")
	ErrRollbackOnly      = errors.New("b5coordinator: transaction is rollback-only")
	ErrTerminal          = errors.New("b5coordinator: transaction is terminal")
	ErrCapabilityUnknown = errors.New("b5coordinator: capability result is uncertain")
)

Functions

func ErrorCode

func ErrorCode(err error) b5.ErrorCode

Types

type Analysis

type Analysis struct {
	Facts          b5dml.StatementFacts
	TreeDigest     [32]byte
	ManifestDigest [32]byte
	ClosureDigest  [32]byte
	Decision       Decision
	EstimatedRows  int64
	EstimatedWork  uint64
	Artifact       any
}

Analysis is immutable candidate-phase evidence. Artifact may contain the binder enrollment used for same-transaction comparison, but it must not be serializable or executable and is never included in a digest.

type Analyzer

type Analyzer interface {
	Analyze(context.Context, StatementRequest, int) (Analysis, error)
}

type ApprovalConsumeRequest

type ApprovalConsumeRequest struct {
	ApprovalID, TransactionID, RequestID string
	PlanDigest                           [32]byte
	OwnerEpoch                           uint64
	Now                                  time.Time
}

type ApprovalLease

type ApprovalLease struct {
	ID                   string
	Generation           uint64
	ApprovedBoundsDigest [32]byte
	ExpiresAt            time.Time
}

type ApprovalStore

type ApprovalStore interface {
	Consume(context.Context, ApprovalConsumeRequest) (ApprovalLease, error)
	MarkBeginFailed(context.Context, ApprovalLease, string) error
	MarkActive(context.Context, ApprovalLease, string) error
}

type AuditEvent

type AuditEvent struct {
	Kind                                AuditKind
	TransactionID, SessionID, RequestID string
	Sequence                            uint64
	PlanDigest                          [32]byte
	StatementOrdinal                    int
	Action                              b5.DMLAction
	AffectedRows                        int64
	DBOutcome                           b5.DBOutcome
	ErrorCode                           b5.ErrorCode
	EvidenceDigest                      [32]byte
}

type AuditKind

type AuditKind string
const (
	AuditTxBegin      AuditKind = "tx_begin"
	AuditStatement    AuditKind = "tx_statement"
	AuditCommitIntent AuditKind = "tx_commit_intent"
	AuditRollback     AuditKind = "tx_rollback"
	AuditTerminal     AuditKind = "tx_terminal_outcome"
)

type AuditResult

type AuditResult struct {
	Durability  b5.AuditDurability
	EventDigest [32]byte
}

type Auditor

type Auditor interface {
	Barrier(context.Context, AuditEvent) (AuditResult, error)
}

type BackendIdentity

type BackendIdentity struct {
	PID                                   int
	SecretDigest                          [32]byte
	StartedAt                             time.Time
	ConnectionGeneration, LeaseGeneration uint64
}

type BeginCapability

type BeginCapability interface {
	BeginNative(context.Context) (b5dml.BeginAttemptEvidence, error)
	FixContext(context.Context, time.Time) error
	BindAndSeal(context.Context, Plan) ([32]byte, error)
	Activate() (Capability, error)
	CleanupBegin(context.Context, *b5terminal.TerminalOwner, b5dml.BeginCleanupDecision) (TerminalResult, error)
}

type BeginRequest

type BeginRequest struct {
	Session                              SessionAuthorization
	TransactionID, RequestID, ApprovalID string
	Plan                                 PlanRequest
}

type Capability

Capability is transaction-bound and accepts only a statement already sealed into the plan. It intentionally has no method taking arbitrary SQL.

type Clock

type Clock interface{ Now() time.Time }

type Config

type Config struct {
	Transactions TransactionStore
	Approvals    ApprovalStore
	Sessions     SessionGate
	Engine       Engine
	Audit        Auditor
	Clock        Clock
}

type Coordinator

type Coordinator struct {
	// contains filtered or unexported fields
}

func New

func New(config Config) (*Coordinator, error)

func (*Coordinator) Begin

func (coordinator *Coordinator) Begin(ctx context.Context, request BeginRequest, analyzer Analyzer) (Result, error)

func (*Coordinator) Commit

func (coordinator *Coordinator) Commit(ctx context.Context, request FinishRequest) (Result, error)

func (*Coordinator) Execute

func (coordinator *Coordinator) Execute(ctx context.Context, request ExecuteRequest) (Result, error)

func (*Coordinator) Expire

func (coordinator *Coordinator) Expire(ctx context.Context, limit int) (int, error)

Expire performs the durable deadline scan used by an owner watchdog. It never touches a row owned by another epoch because only locally-held live capabilities can be terminated; other rows remain for the owner-crash reaper/reconciliation path.

func (*Coordinator) FinalFence

func (coordinator *Coordinator) FinalFence(ctx context.Context, transactionID string) error

FinalFence durably marks that all post-terminal control-plane work for a transaction completed. It is deliberately callable only after the business connection has reached a terminal disposition.

func (*Coordinator) Recover

func (coordinator *Coordinator) Recover(ctx context.Context, transactionID string) (Result, error)

func (*Coordinator) Rollback

func (coordinator *Coordinator) Rollback(ctx context.Context, request FinishRequest) (Result, error)

func (*Coordinator) Shutdown

func (coordinator *Coordinator) Shutdown(ctx context.Context) error

Shutdown stops admission at the protocol owner, freezes every locally-held capability, and gives each transaction the same typed rollback/cancel-taint lifecycle used by the watchdog. It never reconstructs or touches a foreign owner capability.

func (*Coordinator) Status

func (coordinator *Coordinator) Status(ctx context.Context, session SessionAuthorization, transactionID string) (Result, error)

func (*Coordinator) String

func (coordinator *Coordinator) String() string

type Decision

type Decision uint8
const (
	DecisionUnknown Decision = iota
	DecisionAllow
	DecisionWarn
	DecisionApprove
	DecisionDeny
	DecisionMask
)

type DirectorySessionGate

type DirectorySessionGate struct{ Directory *b5session.Directory }

DirectorySessionGate preserves b5session's proof/owner/sticky-route rules. It performs Lookup, never Touch: invalid, stale and wrong-instance requests therefore cannot extend a victim transaction or acquire cleanup authority.

func (DirectorySessionGate) Validate

func (gate DirectorySessionGate) Validate(ctx context.Context, authorization SessionAuthorization) error

type Engine

type Engine interface {
	Pin(context.Context, Plan, *b5dml.BeginMachine) (BeginCapability, BackendIdentity, error)
}

type ExecuteRequest

type ExecuteRequest struct {
	Session                               SessionAuthorization
	TransactionID, RequestID, OperationID string
	Ordinal                               int
}

type Failure

type Failure struct {
	Code  b5.ErrorCode
	Cause error
}

func (*Failure) Error

func (failure *Failure) Error() string

func (*Failure) Unwrap

func (failure *Failure) Unwrap() error

type FinishRequest

type FinishRequest struct {
	Session                  SessionAuthorization
	TransactionID, RequestID string
}

type MemoryApproval

type MemoryApproval struct {
	ID            string
	PlanDigest    [32]byte
	BoundsDigest  [32]byte
	ExpiresAt     time.Time
	Status        string
	Generation    uint64
	TransactionID string
}

type MemoryApprovalStore

type MemoryApprovalStore struct {
	// contains filtered or unexported fields
}

func NewMemoryApprovalStore

func NewMemoryApprovalStore() *MemoryApprovalStore

func (*MemoryApprovalStore) Consume

func (*MemoryApprovalStore) Get

func (memory *MemoryApprovalStore) Get(id string) (MemoryApproval, bool)

func (*MemoryApprovalStore) MarkActive

func (memory *MemoryApprovalStore) MarkActive(_ context.Context, lease ApprovalLease, transactionID string) error

func (*MemoryApprovalStore) MarkBeginFailed

func (memory *MemoryApprovalStore) MarkBeginFailed(_ context.Context, lease ApprovalLease, transactionID string) error

func (*MemoryApprovalStore) Put

func (memory *MemoryApprovalStore) Put(value MemoryApproval)

type MemoryAuditor

type MemoryAuditor struct {
	Events     []AuditEvent
	Fail       func(AuditEvent) error
	Durability func(AuditEvent) b5.AuditDurability
	// contains filtered or unexported fields
}

func (*MemoryAuditor) Barrier

func (audit *MemoryAuditor) Barrier(_ context.Context, event AuditEvent) (AuditResult, error)

type MemoryTransactionStore

type MemoryTransactionStore struct {
	// contains filtered or unexported fields
}

MemoryTransactionStore is a deterministic implementation of the exact durable-store contract. It is intended for coordinator model/fault tests; production construction uses store.B5TransactionRepository.

func NewMemoryTransactionStore

func NewMemoryTransactionStore() *MemoryTransactionStore

func (*MemoryTransactionStore) CASProgress

func (*MemoryTransactionStore) CASState

func (memory *MemoryTransactionStore) CASState(_ context.Context, id string, revision int64, fromStatus b5.TransactionStatus, fromPhase b5.TransactionPhase, toStatus b5.TransactionStatus, toPhase b5.TransactionPhase) (store.B5Transaction, error)

func (*MemoryTransactionStore) Create

func (*MemoryTransactionStore) Get

func (*MemoryTransactionStore) ListExpired

func (memory *MemoryTransactionStore) ListExpired(_ context.Context, now time.Time, limit int) ([]store.B5Transaction, error)

type OperationEvidence

type OperationEvidence struct {
	Ordinal                 int      `json:"ordinal"`
	OperationID             string   `json:"operation_id"`
	AffectedRows            int64    `json:"affected_rows"`
	StatementEvidenceDigest [32]byte `json:"statement_evidence_digest"`
	AuditEventDigest        [32]byte `json:"audit_event_digest"`
}

type Plan

type Plan struct {
	SchemaID, TenantID, PrincipalID, AgentID, DatasourceID string
	SchemaVersion                                          uint16
	KeyRevision, DatasourceRevision, PolicyRevision        uint64
	Dialect                                                string
	ServerMajor                                            int
	Isolation, BinderABI, ClosurePolicy                    string
	Limits                                                 ResourceLimits
	Statements                                             []PlannedStatement
	Digest                                                 [32]byte
	RequiresApproval                                       bool
}

func Preflight

func Preflight(ctx context.Context, request PlanRequest, analyzer Analyzer) (Plan, error)

Preflight analyzes the complete ordered plan before the coordinator is able to ask Engine for a connection. A single invalid operation rejects the whole plan and no partially analyzed plan is returned.

type PlanAnalyzerResolver

type PlanAnalyzerResolver interface {
	AnalyzerFor(context.Context, PlanRequest) (Analyzer, error)
}

PlanAnalyzerResolver selects an analyzer from server-owned datasource and principal state. It is evaluated before any statement analysis and prevents a multi-datasource production runtime from trusting client dialect metadata.

type PlanRequest

type PlanRequest struct {
	TenantID, PrincipalID, AgentID, DatasourceID    string
	KeyRevision, DatasourceRevision, PolicyRevision uint64
	Dialect                                         string
	ServerMajor                                     int
	Isolation, BinderABI, ClosurePolicy             string
	Statements                                      []StatementRequest
	Limits                                          ResourceLimits
}

type PlannedStatement

type PlannedStatement struct {
	Ordinal        int
	OperationID    string
	SQL            string
	RawSQLDigest   [32]byte
	ReasonDigest   [32]byte
	TreeDigest     [32]byte
	ManifestDigest [32]byte
	ClosureDigest  [32]byte
	Action         b5.DMLAction
	Writes         []b5dml.WriteTarget
	References     []b5dml.Reference
	Decision       Decision
	EstimatedRows  int64
	EstimatedWork  uint64
	// contains filtered or unexported fields
}

func (PlannedStatement) Artifact

func (statement PlannedStatement) Artifact() any

type ResourceLimits

type ResourceLimits struct {
	MaxStatements     int
	MaxSQLBytes       int
	MaxPlanBytes      int
	MaxAffectedRows   int64
	MaxEstimatedWork  uint64
	IdleTimeout       time.Duration
	WallTimeout       time.Duration
	StatementTimeout  time.Duration
	OperationWatchdog time.Duration
	QuiesceGrace      time.Duration
}

func DefaultResourceLimits

func DefaultResourceLimits() ResourceLimits

type Result

type Result struct {
	TransactionID         string                   `json:"transaction_id"`
	Status                b5.TransactionStatus     `json:"status"`
	Phase                 b5.TransactionPhase      `json:"phase"`
	Code                  b5.ErrorCode             `json:"error_code,omitempty"`
	Effect                b5.TxEffect              `json:"tx_effect"`
	DBOutcome             b5.DBOutcome             `json:"db_outcome,omitempty"`
	AuditDurability       b5.AuditDurability       `json:"audit_durability,omitempty"`
	ConnectionDisposition b5.ConnectionDisposition `json:"connection_disposition,omitempty"`
	NextOrdinal           int                      `json:"next_ordinal"`
	AffectedRows          int64                    `json:"affected_rows"`
	Evidence              []OperationEvidence      `json:"evidence,omitempty"`
}

type SessionAuthorization

type SessionAuthorization struct {
	SessionID, McpSessionID, PrincipalID, InstanceID string
	Method, RequestID, Proof                         string
	OwnerEpoch                                       uint64
	ExpectedSeq                                      *uint64
	BodyDigest                                       [32]byte
}

type SessionFailure

type SessionFailure struct{ Code b5.ErrorCode }

func (*SessionFailure) Error

func (failure *SessionFailure) Error() string

type SessionGate

type SessionGate interface {
	Validate(context.Context, SessionAuthorization) error
}

type StatementRequest

type StatementRequest struct {
	OperationID string
	SQL         string
	Reason      string
}

type StatementResult

type StatementResult struct {
	AffectedRows      int64
	EvidenceDigest    [32]byte
	Decision          Decision
	CapabilityCertain bool
}

type StaticSessionGate

type StaticSessionGate struct {
	SessionID  string
	OwnerEpoch uint64
}

func (StaticSessionGate) Validate

func (gate StaticSessionGate) Validate(_ context.Context, session SessionAuthorization) error

type TerminalResult

type TerminalResult struct {
	Resolution             b5terminal.TerminalResolution
	EvidenceDigest         [32]byte
	DispositionProofDigest [32]byte
	CancelEmission         b5terminal.CancelEmission
}

type TransactionStore

type TransactionStore interface {
	store.B5TransactionStore
}

type WALAuditor

type WALAuditor struct {
	Service *b5wal.Service
	// contains filtered or unexported fields
}

WALAuditor adapts the S7b primary/WAL/receipt service to coordinator barriers. Pre-effect tx_begin is primary-only and fails closed. Post-effect facts use Service so a primary outage is represented by an immutable WAL append receipt and the returned durability axis.

func NewWALAuditor

func NewWALAuditor(service *b5wal.Service) (*WALAuditor, error)

func (*WALAuditor) Barrier

func (audit *WALAuditor) Barrier(ctx context.Context, event AuditEvent) (AuditResult, error)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL