b5wal

type AppendConfirmationState ¶

type AppendConfirmationState = b5.AppendConfirmation

type AppendIdentity ¶

type AppendIdentity struct {
	SegmentID     [16]byte
	RecordOrdinal uint64
	Nonce         [12]byte
	EventUUID     [16]byte
	EventDigest   [32]byte
	ExtentDigest  [32]byte
}

type AppendResult ¶

type AppendResult struct {
	Identity AppendIdentity
	Late     <-chan error
}

AppendResult returns a buffered Late channel only on timeout. Exactly one fsync result will arrive. A late success never reopens this writer.

type ContextKeyWrapper ¶

type ContextKeyWrapper interface {
	WrapSegmentKey(ctx context.Context, keyID string, segmentID [16]byte, segmentKey [32]byte) ([]byte, error)
}

ContextKeyWrapper is the production envelope-encryption surface. The legacy KeyWrapper method remains for isolated S7a tests; production wrappers bind ciphertext to the immutable key and segment identity as authenticated data.

type DBOutcome ¶

type DBOutcome string

const (
	OutcomeNotCommitted DBOutcome = "NOT_COMMITTED"
	OutcomeCommitted    DBOutcome = "COMMITTED"
	OutcomeUnknown      DBOutcome = "UNKNOWN"
)

type DecodedRecord ¶

type DecodedRecord struct {
	Record
	EventPayloadDigest [32]byte
	FrameLength        uint32
	ChargedBytes       uint32
}

DecodedRecord is returned only after length, commit marker, CRC32C, AEAD, event digest, zero padding, and accounting metadata have all verified.

func DecodeRecord(key, extent []byte) (DecodedRecord, error)

type DurabilityAttestor ¶

type DurabilityAttestor interface {
	Attest(context.Context, string) error
}

DurabilityAttestor represents the deployment-specific destructive power-loss qualification that fsync alone cannot prove (for example, volatile device caches that acknowledge barriers). A production file manager is not created unless the target volume's attestation is currently valid.

type EmissionPath ¶

type EmissionPath struct {
	State                            LastSuccessState
	Transition                       Transition
	Records                          []RecordKind
	Rotation                         bool
	PermanentLossIfReplacementWedges []RecordKind
}

EmissionPath is one complete path after the named primary last-success point. Multiplicity is len(Records): each semantic kind is emitted at most once. A post-timeout rotation creates no record, the diagnostic coalesces all failures for the reservation, and recovery is a single coalesced record.

func EmissionPaths() []EmissionPath

type EmissionReservation ¶

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

EmissionReservation combines the executable emission path with the physical 120KiB reservation. Facts must arrive once and in frozen table order.

func NewEmissionReservation(id [16]byte, state LastSuccessState, transition Transition) (*EmissionReservation, error)

func (reservation *EmissionReservation) Emitted() int

func (reservation *EmissionReservation) RemainingBytes() int

type EnvelopeKMS ¶

type EnvelopeKMS interface {
	Encrypt(ctx context.Context, masterRevision string, plaintext, aad []byte) ([]byte, error)
	Decrypt(ctx context.Context, masterRevision string, ciphertext, aad []byte) ([]byte, error)
}

EnvelopeKMS is intentionally independent of the B3 redaction registry. It follows the same revisioned-key/envelope style but never shares key material or activation state with redaction keys.

type Extension ¶

type Extension struct {
	Tag   uint16
	Value []byte
}

Extension is a canonical binary header extension. Tags must be non-zero, strictly increasing, and unique. Each entry is encoded as uint16 tag, uint16 value length, then value bytes. The aggregate encoding is capped at MaxHeaderExtension bytes and unused header bytes are zero.

type ExtentSink ¶

type ExtentSink interface {
	WriteExtent(extent []byte) error
	Sync() error
	Close() error
}

ExtentSink writes exactly one aligned extent and durably syncs it.

type FileExtentSink ¶

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

func NewFileExtentSink(path string) (*FileExtentSink, error)

func (sink *FileExtentSink) Close() error

func (sink *FileExtentSink) Sync() error

func (sink *FileExtentSink) WriteExtent(extent []byte) error

type FileKeyRegistry ¶

type FileKeyRegistry struct{ Directory string }

FileKeyRegistry is a durable, process-shared CAS registry for a WAL volume. Create-exclusive files provide the CAS; file and parent directory fsync make a successful reservation survive restart. The registry stores only a digest of KMS ciphertext, never plaintext segment keys.

func (registry FileKeyRegistry) Reserve(ctx context.Context, keyID string, segmentID [16]byte, wrappedKey []byte) (bool, error)

type FileManifestStore ¶

type FileManifestStore struct{ Directory string }

FileManifestStore is the local manifest implementation. Each manifest is create-exclusive, mode 0600, file-synced, then its parent directory is synced.

func (store FileManifestStore) Persist(ctx context.Context, manifest SegmentManifest) error

type FileQuarantineStore ¶

type FileQuarantineStore struct{ Directory string }

func (store FileQuarantineStore) Quarantine(ctx context.Context, source string, cause error) error

type FileScanner ¶

type FileScanner struct {
	Directory  string
	Unwrapper  SegmentKeyUnwrapper
	Quarantine QuarantineStore
}

FileScanner accepts only complete authenticated extents. A torn tail, manifest mismatch, ordinal gap/reorder, bad AEAD/digest, or duplicate nonce quarantines the whole segment; no prefix is replayed from that segment.

func (scanner FileScanner) Scan(ctx context.Context) (ScanResult, error)

type GlobalEntry ¶

type GlobalEntry struct {
	Position       uint64
	PreviousDigest [32]byte
	Digest         [32]byte
	EventUUID      [16]byte
	PayloadDigest  [32]byte
}

type HistoricalResponseState ¶

type HistoricalResponseState uint8

const (
	HistoricalResponseUnknown HistoricalResponseState = iota
	HistoricalResponsePrepared
)

type IdentityKeyWrapper ¶

type IdentityKeyWrapper struct{}

IdentityKeyWrapper is only a test implementation. Production wiring must provide an authenticated KMS-backed wrapper before this package is enabled.

func (IdentityKeyWrapper) Wrap(_ context.Context, segmentKey [32]byte) ([]byte, error)

type IngestResult ¶

type IngestResult struct {
	Durability ReportedDurability
	Append     AppendResult
	Primary    bool
}

type KMSKeyWrapper ¶

type KMSKeyWrapper struct {
	KMS            EnvelopeKMS
	MasterRevision string
}

KMSKeyWrapper wraps one independently generated key per segment. A KMS outage is returned as ErrKeyUnavailable; callers must fail closed.

func (wrapper KMSKeyWrapper) UnwrapSegmentKey(ctx context.Context, manifest SegmentManifest) ([32]byte, error)

func (wrapper KMSKeyWrapper) Wrap(context.Context, [32]byte) ([]byte, error)

func (wrapper KMSKeyWrapper) WrapSegmentKey(ctx context.Context, keyID string, segmentID [16]byte, key [32]byte) ([]byte, error)

type KeyRegistry ¶

type KeyRegistry interface {
	Reserve(ctx context.Context, keyID string, segmentID [16]byte, wrappedKey []byte) (bool, error)
}

KeyRegistry reserves key identities by compare-and-set. A false result means the key ID already exists; callers must discard the candidate key and ID.

type KeyWrapper ¶

type KeyWrapper interface {
	Wrap(ctx context.Context, segmentKey [32]byte) ([]byte, error)
}

KeyWrapper represents the external master-key operation. The raw segment key is never written to the manifest by SegmentFactory.

type LastSuccessState ¶

type LastSuccessState string

const (
	LastBeforeBeginBarrier      LastSuccessState = "before_begin_barrier"
	LastActiveBeforeEffect      LastSuccessState = "active_before_statement_effect"
	LastAfterStatementEffect    LastSuccessState = "after_statement_effect"
	LastRollbackOnly            LastSuccessState = "rollback_only"
	LastBeforeCommitIntent      LastSuccessState = "before_commit_intent_durable"
	LastIntentDurableBeforeSend LastSuccessState = "intent_durable_before_commit_send"
	LastPositiveCommitACK       LastSuccessState = "positive_commit_ack"
	LastUnknownCommitOutcome    LastSuccessState = "unknown_commit_outcome"
	LastRollbackACK             LastSuccessState = "rollback_ack"
)

type Manager ¶

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

Manager creates a fresh segment on startup and supports at most one rotation after a wedge for each reservation. Rotation itself emits no WAL record. If the replacement segment wedges on the same reservation, no second rotation, diagnostic, or retry is allowed; remaining facts are LOST and are represented only by the durable result receipt plane.

func NewFileManager(ctx context.Context, factory SegmentFactory, owner uint64, directory string, attestor DurabilityAttestor) (*Manager, error)

func NewManager(ctx context.Context, factory SegmentFactory, owner uint64, newSink func(Segment) (ExtentSink, error)) (*Manager, error)

func (manager *Manager) ChangeOwnership(ctx context.Context, owner uint64) error

func (manager *Manager) Close() error

func (manager *Manager) Current() *Writer

func (manager *Manager) RotateAfterWedge(ctx context.Context, reservationID [16]byte) error

type ManifestStore ¶

type ManifestStore interface {
	Persist(ctx context.Context, manifest SegmentManifest) error
}

ManifestStore must make the manifest durable, including its parent directory, before Persist returns nil.

type MemoryKeyRegistry ¶

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

MemoryKeyRegistry is a concurrency-safe contract implementation for tests. It models durable CAS state only; it is not a production registry.

func NewMemoryKeyRegistry() *MemoryKeyRegistry

func (registry *MemoryKeyRegistry) Reserve(_ context.Context, keyID string, segmentID [16]byte, _ []byte) (bool, error)

type MemoryManifestStore ¶

type MemoryManifestStore struct {
	Manifests []SegmentManifest
	Err       error
	// contains filtered or unexported fields
}

MemoryManifestStore is a test implementation that records the persist order.

func (store *MemoryManifestStore) Persist(_ context.Context, manifest SegmentManifest) error

type MemoryReceiptStore ¶

type MemoryReceiptStore struct {
	PutErr     error
	AdvanceErr error
	// contains filtered or unexported fields
}

func NewMemoryReceiptStore() *MemoryReceiptStore

func (store *MemoryReceiptStore) Advance(_ context.Context, key ReceiptKey, update ReceiptAdvance) (ResultReceipt, error)

func (store *MemoryReceiptStore) FindByEventUUID(_ context.Context, eventUUID [16]byte) ([]ResultReceipt, error)

func (store *MemoryReceiptStore) Get(_ context.Context, key ReceiptKey) (ResultReceipt, bool, error)

func (store *MemoryReceiptStore) PutWriteOnce(_ context.Context, receipt ResultReceipt) (ResultReceipt, bool, error)

type PGPrimaryStore ¶

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

PGPrimaryStore writes canonical v4 facts into the frozen b5_tx_events table. It is named for production use but is dialect-neutral for SQLite tests.

func NewPGPrimaryStore(repository store.B5TxEventStore) (*PGPrimaryStore, error)

func (primary *PGPrimaryStore) Put(ctx context.Context, event PrimaryEvent) (bool, error)

func (primary *PGPrimaryStore) Snapshot(ctx context.Context, limit int) ([]PrimaryEvent, error)

type PGReceiptStore ¶

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

PGReceiptStore adapts the frozen b5_result_receipts repository to the WAL receipt contract. Despite the name it deliberately works with the SQLite repository too, so the exact persistence semantics can be tested cheaply. Production construction supplies Store.B5ResultReceipts(), backed by PG.

func NewPGReceiptStore(repository store.B5ResultReceiptStore) (*PGReceiptStore, error)

func (adapter *PGReceiptStore) Advance(ctx context.Context, key ReceiptKey, update ReceiptAdvance) (ResultReceipt, error)

func (adapter *PGReceiptStore) FindByEventUUID(ctx context.Context, eventUUID [16]byte) ([]ResultReceipt, error)

func (adapter *PGReceiptStore) Get(ctx context.Context, key ReceiptKey) (ResultReceipt, bool, error)

func (adapter *PGReceiptStore) PutWriteOnce(ctx context.Context, receipt ResultReceipt) (ResultReceipt, bool, error)

type PrimaryEvent ¶

type PrimaryEvent struct {
	ReplayEvent
	Kind           RecordKind
	Outcome        DBOutcome
	SchemaID       string
	SchemaVersion  uint16
	CanonicalEvent []byte
}

PrimaryEvent is the already-canonical fact presented to the durable primary audit plane. Put must be idempotent for equal UUID+PayloadDigest and fail closed with ErrEventUUIDConflict for an unequal digest.

type PrimarySnapshot ¶

type PrimarySnapshot interface {
	Snapshot(context.Context, int) ([]PrimaryEvent, error)
}

type PrimaryStore ¶

type PrimaryStore interface {
	Put(context.Context, PrimaryEvent) (alreadyPresent bool, err error)
}

type QuarantineStore ¶

type QuarantineStore interface {
	Quarantine(context.Context, string, error) error
}

type ReceiptAdvance ¶

type ReceiptAdvance struct {
	Append         *AppendConfirmationState
	Reconciliation *ReconciliationState
	Delivery       *ResponseDeliveryState
}

type ReceiptFinder ¶

type ReceiptFinder interface {
	FindByEventUUID(context.Context, [16]byte) ([]ResultReceipt, error)
}

type ReceiptKey ¶

type ReceiptKey struct {
	SessionID         string
	RequestID         string
	EventUUID         [16]byte
	AttemptGeneration uint64
}

type ReceiptStore ¶

type ReceiptStore interface {
	PutWriteOnce(ctx context.Context, receipt ResultReceipt) (ResultReceipt, bool, error)
	Get(ctx context.Context, key ReceiptKey) (ResultReceipt, bool, error)
	Advance(ctx context.Context, key ReceiptKey, update ReceiptAdvance) (ResultReceipt, error)
}

type Reconciler ¶

type Reconciler struct {
	Primary  PrimaryStore
	Receipts ReceiptStore
	Finder   ReceiptFinder
}

Reconciler replays verified records in double-chain order and durably joins them to the result-receipt plane. The returned ledger is an operational view; receipt updates and primary rows are the durable recovery ledger.

func (reconciler Reconciler) Reconcile(ctx context.Context, records []RecoveredRecord) ([]RecoveryLedgerEntry, error)

type ReconciliationState ¶

type ReconciliationState = b5.Reconciliation

type Record ¶

type Record struct {
	SegmentID          [16]byte
	RecordOrdinal      uint64
	ReservationID      [16]byte
	EventUUID          [16]byte
	EventSchemaID      string
	EventSchemaVersion uint16
	KeyID              string
	Nonce              [12]byte
	CanonicalEvent     []byte
	Extensions         []Extension
}

Record identifies and carries one already-canonical event-v4 payload.

type RecordKind ¶

type RecordKind string

RecordKind is an event-v4 semantic kind. Result receipts are deliberately absent: they live in ReceiptStore and do not consume WAL record slots.

const (
	RecordStatement            RecordKind = "tx_statement"
	RecordRollback             RecordKind = "tx_rollback"
	RecordTerminalOutcome      RecordKind = "tx_terminal_outcome"
	RecordSessionTerminal      RecordKind = "session_terminal"
	RecordDurabilityDiagnostic RecordKind = "durability_diagnostic"
	RecordRecovery             RecordKind = "tx_durability_recovered"
	RecordCommitIntent         RecordKind = "tx_commit_intent"
)

type RecoveredRecord ¶

type RecoveredRecord struct {
	Decoded  DecodedRecord
	Identity AppendIdentity
	Event    PrimaryEvent
	Source   string
}

type RecoveryClassification ¶

type RecoveryClassification string

const (
	RecoveryHistoricalUnknown        RecoveryClassification = "HISTORICAL_RESPONSE_UNKNOWN"
	RecoveryCommittedAuditPending    RecoveryClassification = "COMMITTED_AUDIT_PENDING"
	RecoveryNotCommittedAuditPending RecoveryClassification = "NOT_COMMITTED_AUDIT_PENDING"
	RecoveryDBOutcomeUnknown         RecoveryClassification = "UNKNOWN"
	RecoveryPrimaryDurable           RecoveryClassification = "PRIMARY_DURABLE"
)

type RecoveryLedgerEntry ¶

type RecoveryLedgerEntry struct {
	EventUUID          [16]byte
	TransactionID      string
	TransactionSeq     uint64
	Classification     RecoveryClassification
	HistoricalResponse HistoricalResponseState
	ReportedDurability ReportedDurability
	AppendConfirmation AppendConfirmationState
	Reconciliation     ReconciliationState
	Replayed           bool
}

type ReplayEvent ¶

type ReplayEvent struct {
	EventUUID        [16]byte
	PayloadDigest    [32]byte
	TransactionID    string
	TransactionSeq   uint64
	PreviousTxDigest [32]byte
}

type ReplayVerifier ¶

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

func NewReplayVerifier() *ReplayVerifier

func (verifier *ReplayVerifier) GlobalChain() []GlobalEntry

func (verifier *ReplayVerifier) Ingest(event ReplayEvent) (GlobalEntry, bool, error)

type ReportedDurability ¶

type ReportedDurability = b5.AuditDurability

type Reservation ¶

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

Reservation accounts physical extents, not plaintext bytes. A normal B5 path is created with MaxPathRecordCount and therefore exactly 120 KiB.

func NewReservation(recordCount int) (*Reservation, error)

func (reservation *Reservation) ConsumeEncodedRecord(extent []byte) error

func (reservation *Reservation) Remaining() int

type ResponseDeliveryState ¶

type ResponseDeliveryState = b5.DeliveryStatus

type RestartView ¶

type RestartView struct {
	HistoricalResponse HistoricalResponseState
	Receipt            ResultReceipt
	RecordValid        bool
}

func RestartJoin(ctx context.Context, store ReceiptStore, key ReceiptKey, recordValid, primaryDurable bool) (RestartView, error)

type ResultReceipt ¶

type ResultReceipt struct {
	SchemaID                     string
	SchemaVersion                uint16
	Key                          ReceiptKey
	BusinessEventDigest          [32]byte
	WALAppendReceiptDigest       [32]byte
	ReportedDurabilityAtResponse ReportedDurability
	AppendConfirmation           AppendConfirmationState
	Reconciliation               ReconciliationState
	Delivery                     ResponseDeliveryState
}

ResultReceipt is write-once for Key, both digests, and ReportedDurabilityAtResponse. Mutable state advances only through the transition functions below. Persisting Prepared before sending makes the exact terminal result reproducible even if the process dies before or during network delivery; SendStarted cannot prove that a peer received the bytes.

type ScanResult ¶

type ScanResult struct {
	Records     []RecoveredRecord
	Quarantined []string
}

type Segment ¶

type Segment struct {
	Manifest SegmentManifest
	Key      [32]byte
}

type SegmentFactory ¶

type SegmentFactory struct {
	Random         io.Reader
	MasterRevision string
	Registry       KeyRegistry
	Wrapper        KeyWrapper
	Manifests      ManifestStore
}

func (factory SegmentFactory) Create(ctx context.Context, ownerGeneration uint64) (Segment, error)

type SegmentKeyUnwrapper ¶

type SegmentKeyUnwrapper interface {
	UnwrapSegmentKey(context.Context, SegmentManifest) ([32]byte, error)
}

type SegmentManifest ¶

type SegmentManifest struct {
	FormatVersion   uint16   `json:"format_version"`
	SegmentID       [16]byte `json:"segment_id"`
	KeyID           string   `json:"key_id"`
	WrappedKey      []byte   `json:"wrapped_key"`
	NonceDomain     [4]byte  `json:"nonce_domain"`
	OwnerGeneration uint64   `json:"owner_generation"`
}

type Service ¶

type Service struct {
	Primary     PrimaryStore
	Receipts    ReceiptStore
	WAL         *Manager
	SyncTimeout time.Duration
}

Service is the flag-off production boundary intended for S6. It owns primary-first ingestion, emergency staging, the response receipt fence, and one post-wedge rotation. It has no dependency on MCP or HTTP packages.

func (service *Service) IngestAndRespond(ctx context.Context, reservation *EmissionReservation, event ServiceEvent, send func(ResultReceipt) error) (IngestResult, error)

type ServiceEvent ¶

type ServiceEvent struct {
	Kind              RecordKind
	Outcome           DBOutcome
	TransactionID     string
	TransactionSeq    uint64
	PreviousTxDigest  [32]byte
	EventUUID         [16]byte
	SchemaID          string
	SchemaVersion     uint16
	CanonicalEvent    []byte
	SessionID         string
	RequestID         string
	AttemptGeneration uint64
}

type Transition ¶

type Transition string

const (
	TransitionPrimaryFailure Transition = "primary_failure"
	TransitionFsyncTimeout   Transition = "wal_fsync_timeout"
	TransitionLateRecovery   Transition = "late_or_restart_recovery"
)

type Writer ¶

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

func NewWriter(segment Segment, sink ExtentSink) (*Writer, error)

func (writer *Writer) Append(ctx context.Context, record Record) (AppendResult, error)

func (writer *Writer) AppendReserved(ctx context.Context, reservation *Reservation, record Record) (AppendResult, error)

func (writer *Writer) Seal() error

func (writer *Writer) Segment() Segment

func (writer *Writer) State() WriterState

type WriterState ¶

type WriterState uint8

const (
	WriterOpen WriterState = iota
	WriterWedgedUnconfirmed
	WriterSealed
)