lifecycle

package
v0.14.0 Latest Latest
Warning

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

 Go to latest Published: Aug 9, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// DefaultMaxMembershipPeers bounds the explicitly configured replica set.
	DefaultMaxMembershipPeers = DefaultMaxRepairPeers
	// DefaultMaxMembershipPeerIDBytes bounds one configured replica identity.
	DefaultMaxMembershipPeerIDBytes = DefaultMaxRepairPeerIDBytes
	// DefaultMaxMembershipBytes bounds the complete membership file envelope.
	DefaultMaxMembershipBytes = DefaultMaxRepairWatermarkBytes
)
View Source 
const (
	DefaultMaxNamespaceBytes   = 128
	DefaultMaxLogicalKeyBytes  = 1024
	DefaultMaxAuthorityIDBytes = 128
	DefaultMaxRecordBytes      = 64 << 10
	DefaultMaxCheckpointBytes  = 64 << 20
	DefaultMaxJournalBytes     = 64 << 20
	DefaultMaxJournalEntries   = 65536
)
View Source 
const (
	// RepairBatchMessageType identifies an ordered journal tail batch.
	RepairBatchMessageType = "lifecycle.repair.batch"
	// RepairSnapshotMessageType identifies a complete checkpoint bootstrap.
	RepairSnapshotMessageType = "lifecycle.repair.snapshot"
	// RepairAckMessageType identifies an acknowledgement of an applied watermark.
	RepairAckMessageType = "lifecycle.repair.ack"
	// DefaultMaxRepairRecords bounds one repair batch or snapshot.
	DefaultMaxRepairRecords = 128
	// DefaultMaxRepairLogicalKeyBytes bounds logical namespace/key bytes in one repair payload.
	DefaultMaxRepairLogicalKeyBytes = 64 << 10
	// DefaultMaxRepairMetadataBytes bounds encoded lifecycle metadata in one repair payload.
	DefaultMaxRepairMetadataBytes = 64 << 10
	// DefaultMaxRepairFrameBytes bounds one repair JSON payload independently from raw blobs.
	DefaultMaxRepairFrameBytes = 128 << 10
)
View Source 
const (
	// LifecycleCapabilityV1 identifies the capability required for lifecycle records.
	LifecycleCapabilityV1 = "lifecycle.v1"
	// LifecycleRepairReconcileCapabilityV1 enables startup watermark reconciliation.
	LifecycleRepairReconcileCapabilityV1 = "lifecycle.repair-reconcile.v1"
	// LifecycleRecordMessageType identifies a logical lifecycle mutation envelope.
	LifecycleRecordMessageType = "lifecycle.record"
	// DefaultMaxLifecyclePayloadBytes bounds the JSON envelope independently from raw blobs.
	DefaultMaxLifecyclePayloadBytes = 128 << 10
)
View Source 
const (
	// DefaultMaxRepairPeers bounds the durable acknowledgement map.
	DefaultMaxRepairPeers = 1024
	// DefaultMaxRepairPeerIDBytes bounds one durable peer identity.
	DefaultMaxRepairPeerIDBytes = 128
	// DefaultMaxRepairWatermarkBytes bounds the complete watermark file envelope.
	DefaultMaxRepairWatermarkBytes = 1 << 20
)
View Source 
const DefaultMaxAuthorityStateBytes = 1 << 20

Variables

View Source 
var (
	// ErrNilApplierState is returned when an applier has no logical state view.
	ErrNilApplierState = errors.New("lifecycle: nil applier state")
	// ErrNilApplierJournal is returned when an applier has no durable journal.
	ErrNilApplierJournal = errors.New("lifecycle: nil applier journal")
	// ErrNilApplierBlobStore is returned when a present record cannot access raw blobs.
	ErrNilApplierBlobStore = errors.New("lifecycle: nil applier blob store")
	// ErrLifecycleBlobMissing is returned when a present record references no available blob.
	ErrLifecycleBlobMissing = errors.New("lifecycle: referenced blob is missing")
)
View Source 
var (
	ErrNilAuthorityPath   = errors.New("lifecycle: empty authority path")
	ErrAuthorityCorrupt   = errors.New("lifecycle: corrupt authority state")
	ErrAuthorityChecksum  = errors.New("lifecycle: authority checksum mismatch")
	ErrAuthorityLimit     = errors.New("lifecycle: authority state limit exceeded")
	ErrAuthorityMismatch  = errors.New("lifecycle: authority identity mismatch")
	ErrAuthorityVersion   = errors.New("lifecycle: invalid authority version")
	ErrAuthorityExhausted = errors.New("lifecycle: authority version space exhausted")
)
View Source 
var (
	ErrJournalClosed          = errors.New("lifecycle: journal closed")
	ErrJournalCorrupt         = errors.New("lifecycle: corrupt journal")
	ErrJournalChecksum        = errors.New("lifecycle: journal checksum mismatch")
	ErrJournalVersionOrder    = errors.New("lifecycle: journal version is not increasing")
	ErrJournalLimit           = errors.New("lifecycle: journal limit exceeded")
	ErrCheckpointCorrupt      = errors.New("lifecycle: corrupt checkpoint")
	ErrCheckpointChecksum     = errors.New("lifecycle: checkpoint checksum mismatch")
	ErrCheckpointTooLarge     = errors.New("lifecycle: checkpoint too large")
	ErrCompactionUnsafe       = errors.New("lifecycle: unsafe compaction")
	ErrCompactionNoProgress   = errors.New("lifecycle: compaction watermark does not advance")
	ErrCompactionWatermark    = errors.New("lifecycle: invalid compaction watermark")
	ErrCompactionBaseMismatch = errors.New("lifecycle: compaction base mismatch")
	ErrPeerBehindWatermark    = errors.New("lifecycle: peer is behind compaction watermark")
	ErrSnapshotStale          = errors.New("lifecycle: snapshot is older than journal")
	ErrNilCheckpointPath      = errors.New("lifecycle: empty checkpoint path")
)
View Source 
var (
	// ErrNilMembershipPath is returned when no durable membership path is configured.
	ErrNilMembershipPath = errors.New("lifecycle: empty membership path")
	// ErrNilMembershipBook is returned when a membership operation has no book.
	ErrNilMembershipBook = errors.New("lifecycle: nil membership book")
	// ErrMembershipNotConfigured is returned when no operator-authored membership exists.
	ErrMembershipNotConfigured = errors.New("lifecycle: membership is not configured")
	// ErrMembershipCorrupt is returned for malformed or structurally invalid state.
	ErrMembershipCorrupt = errors.New("lifecycle: corrupt membership")
	// ErrMembershipChecksum is returned when durable membership state fails its CRC.
	ErrMembershipChecksum = errors.New("lifecycle: membership checksum mismatch")
	// ErrMembershipLimit is returned when the membership set or encoded state exceeds bounds.
	ErrMembershipLimit = errors.New("lifecycle: membership limit exceeded")
	// ErrMembershipPeerInvalid is returned for an empty, invalid, or oversized identity.
	ErrMembershipPeerInvalid = errors.New("lifecycle: invalid membership peer identity")
	// ErrMembershipDuplicate is returned when an operator configures one identity twice.
	ErrMembershipDuplicate = errors.New("lifecycle: duplicate membership peer identity")
	// ErrMembershipWatermarkInvalid is returned for an unusable compaction target.
	ErrMembershipWatermarkInvalid = errors.New("lifecycle: invalid membership watermark")
	// ErrMembershipBehind is returned when a configured member lacks sufficient evidence.
	ErrMembershipBehind = errors.New("lifecycle: membership peer is behind compaction watermark")
)
View Source 
var (
	ErrNamespaceEmpty        = errors.New("lifecycle: empty namespace")
	ErrNamespaceTooLarge     = errors.New("lifecycle: namespace too large")
	ErrLogicalKeyEmpty       = errors.New("lifecycle: empty logical key")
	ErrLogicalKeyTooLarge    = errors.New("lifecycle: logical key too large")
	ErrAuthorityEmpty        = errors.New("lifecycle: empty authority")
	ErrAuthorityTooLarge     = errors.New("lifecycle: authority too large")
	ErrAuthorityNotPrintable = errors.New("lifecycle: authority is not printable")
	ErrInvalidState          = errors.New("lifecycle: invalid state")
	ErrInvalidBlobKey        = errors.New("lifecycle: invalid blob key")
	ErrZeroVersion           = errors.New("lifecycle: zero version")
	ErrRecordTooLarge        = errors.New("lifecycle: record too large")
	ErrConflict              = errors.New("lifecycle: same-version record conflict")
	ErrDuplicateLogicalKey   = errors.New("lifecycle: duplicate logical key")
	ErrNilApply              = errors.New("lifecycle: nil apply function")
)
View Source 
var (
	// ErrNilRepairJournal is returned when a repair planner has no journal.
	ErrNilRepairJournal = errors.New("lifecycle: nil repair journal")
	// ErrRepairLimit is returned when a repair batch exceeds its entry or metadata bounds.
	ErrRepairLimit = errors.New("lifecycle: repair limit exceeded")
	// ErrRepairFrameTooLarge is returned when a repair payload exceeds its frame bound.
	ErrRepairFrameTooLarge = errors.New("lifecycle: repair frame too large")
	// ErrRepairMalformed is returned for invalid repair JSON or envelope structure.
	ErrRepairMalformed = errors.New("lifecycle: malformed repair envelope")
	// ErrRepairMessageType is returned for a repair envelope of another kind.
	ErrRepairMessageType = errors.New("lifecycle: unexpected repair message type")
	// ErrRepairOrder is returned when a batch is not strictly ordered after its watermark.
	ErrRepairOrder = errors.New("lifecycle: repair records are not ordered")
	// ErrRepairWatermarkMismatch is returned for a gap or inconsistent batch endpoint.
	ErrRepairWatermarkMismatch = errors.New("lifecycle: repair watermark mismatch")
	// ErrRepairSnapshotRequired is returned when the journal floor is newer than a peer.
	ErrRepairSnapshotRequired = errors.New("lifecycle: repair snapshot required")
	// ErrRepairSnapshotWatermark is returned when a fallback checkpoint does not match the floor.
	ErrRepairSnapshotWatermark = errors.New("lifecycle: repair snapshot watermark mismatch")
)
View Source 
var (
	// ErrNilRepairSessionPeer is returned when a session has no frame peer.
	ErrNilRepairSessionPeer = errors.New("lifecycle: nil repair session peer")
	// ErrNilRepairSessionCoordinator is returned when a session has no planner.
	ErrNilRepairSessionCoordinator = errors.New("lifecycle: nil repair session coordinator")
	// ErrNilRepairSessionApplier is returned when a session receives state without an applier.
	ErrNilRepairSessionApplier = errors.New("lifecycle: nil repair session applier")
)
View Source 
var (
	// ErrLifecyclePayloadTooLarge is returned before decoding an oversized envelope.
	ErrLifecyclePayloadTooLarge = errors.New("lifecycle: transport payload too large")
	// ErrLifecycleEnvelopeMalformed is returned for invalid JSON or invalid record envelopes.
	ErrLifecycleEnvelopeMalformed = errors.New("lifecycle: malformed transport envelope")
	// ErrLifecycleEnvelopeTrailingData is returned when one payload contains more than one JSON value.
	ErrLifecycleEnvelopeTrailingData = errors.New("lifecycle: trailing transport envelope data")
	// ErrLifecycleMessageType is returned when a lifecycle decoder receives another protocol message.
	ErrLifecycleMessageType = errors.New("lifecycle: unexpected transport message type")
	// ErrLifecycleCapabilityRequired is returned before decoding when a peer is not lifecycle-ready.
	ErrLifecycleCapabilityRequired = errors.New("lifecycle: capability is required")
)
View Source 
var (
	// ErrNilWatermarkPath is returned when no durable acknowledgement path is configured.
	ErrNilWatermarkPath = errors.New("lifecycle: empty watermark path")
	// ErrNilWatermarkBook is returned when a watermark operation has no book.
	ErrNilWatermarkBook = errors.New("lifecycle: nil watermark book")
	// ErrWatermarkCorrupt is returned for malformed or structurally invalid state.
	ErrWatermarkCorrupt = errors.New("lifecycle: corrupt repair watermarks")
	// ErrWatermarkChecksum is returned when durable acknowledgement state fails its CRC.
	ErrWatermarkChecksum = errors.New("lifecycle: repair watermark checksum mismatch")
	// ErrWatermarkLimit is returned when the peer map or encoded state exceeds its bounds.
	ErrWatermarkLimit = errors.New("lifecycle: repair watermark limit exceeded")
	// ErrWatermarkPeerInvalid is returned for an empty, invalid, or oversized peer ID.
	ErrWatermarkPeerInvalid = errors.New("lifecycle: invalid repair peer identity")
	// ErrWatermarkRegression is returned when an acknowledgement moves backward.
	ErrWatermarkRegression = errors.New("lifecycle: repair watermark regressed")
	// ErrNilRepairWatermarks is returned when a coordinator has no durable watermark book.
	ErrNilRepairWatermarks = errors.New("lifecycle: nil repair watermarks")
	// ErrRepairAcknowledgement is returned when a peer acknowledges beyond the local journal.
	ErrRepairAcknowledgement = errors.New("lifecycle: repair acknowledgement exceeds journal")
)

Functions

func EncodeRecord 

func EncodeRecord(record Record, limits TransportLimits) ([]byte, error)

EncodeRecord validates and encodes one lifecycle record without applying it.

func EncodeRepairAck 

func EncodeRepairAck(ack RepairAck, limits RepairLimits) ([]byte, error)

EncodeRepairAck validates and encodes one durable watermark acknowledgement.

func EncodeRepairBatch 

func EncodeRepairBatch(batch RepairBatch, limits RepairLimits) ([]byte, error)

EncodeRepairBatch validates and encodes an ordered journal batch.

func EncodeRepairFrame 

func EncodeRepairFrame(frame RepairFrame, limits RepairLimits) ([]byte, error)

EncodeRepairFrame encodes exactly one lifecycle repair message.

func EncodeRepairSnapshot 

func EncodeRepairSnapshot(snapshot RepairSnapshot, limits RepairLimits) ([]byte, error)

EncodeRepairSnapshot validates and encodes a complete checkpoint bootstrap.

func HasLifecycleCapability 

func HasLifecycleCapability(capabilities []string) bool

HasLifecycleCapability reports whether a peer negotiated lifecycle.v1.

func HasRepairReconciliationCapability 

func HasRepairReconciliationCapability(capabilities []string) bool

HasRepairReconciliationCapability reports whether a peer supports startup watermark reconciliation for lifecycle repair sessions.

func OpenJournal 

func OpenJournal(path string, options JournalOptions) (*Journal, Recovery, error)

OpenJournal opens or creates a checksummed journal. A short final envelope is truncated and reported as a recoverable tail; complete corruption fails closed.

func SaveCheckpoint 

func SaveCheckpoint(ctx context.Context, path string, checkpoint Checkpoint, limits Limits) error

SaveCheckpoint writes a validated checkpoint through a durable temp-file rename.

Types

type Applier 

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

Applier makes one capability-gated lifecycle transition durable and deterministic.

func NewApplier 

func NewApplier(blobs storage.BlobStore, state *Store, journal *Journal, limits Limits) (*Applier, error)

NewApplier constructs an applier backed by raw blobs, logical state, and a durable journal.

func (*Applier) Apply 

func (a *Applier) Apply(ctx context.Context, capabilities []string, record Record, suppliedBlob []byte) (ApplyResult, error)

Apply verifies capability and blob durability, appends the journal, then publishes state. suppliedBlob is optional for present records; nil reads the referenced blob from the store.

func (*Applier) ApplySnapshot 

func (a *Applier) ApplySnapshot(ctx context.Context, capabilities []string, snapshot Checkpoint, checkpointPath string) error

ApplySnapshot verifies every referenced blob, durably installs the checkpoint, then publishes the complete logical view. The watermark acknowledgement must be sent by the caller only after this method returns successfully.

type ApplyOutcome 

type ApplyOutcome string

ApplyOutcome classifies a record application. 

const (
	OutcomeApplied   ApplyOutcome = "applied"
	OutcomeDuplicate ApplyOutcome = "duplicate"
	OutcomeStale     ApplyOutcome = "stale"
)

type ApplyResult 

type ApplyResult struct {
	Outcome ApplyOutcome
	Version Version
}

ApplyResult reports the deterministic outcome for one record.

type Authority 

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

Authority durably allocates one total-order version stream for a configured operator-fenced authority identity. A failed mutation may consume a token; it must never be reused.

func OpenAuthority 

func OpenAuthority(ctx context.Context, path, authorityID string, options AuthorityOptions) (*Authority, error)

OpenAuthority opens or creates the durable local allocator. The observed version is adopted when it is newer than the stored allocator state.

func (*Authority) AuthorityID 

func (a *Authority) AuthorityID() string

AuthorityID returns the configured identity that owns this version stream.

func (*Authority) Current 

func (a *Authority) Current() Version

Current returns the last durably allocated token, or the initial epoch with sequence zero when the allocator has not issued a mutation yet.

func (*Authority) Next 

func (a *Authority) Next(ctx context.Context, observed Version) (Version, error)

Next durably allocates a token strictly after both the stored token and the observed journal tail.

type AuthorityOptions 

type AuthorityOptions struct {
	Limits   Limits
	MaxBytes int
	Observed Version
}

AuthorityOptions bounds durable local version state and lets recovery adopt a newer journal tail before the next local mutation is allocated.

type Checkpoint 

type Checkpoint struct {
	Watermark Version  `json:"watermark"`
	Records   []Record `json:"records"`
}

Checkpoint is a complete logical state at a journal watermark.

func LoadCheckpoint 

func LoadCheckpoint(ctx context.Context, path string, limits Limits) (Checkpoint, error)

LoadCheckpoint reads and validates one checkpoint envelope.

type CompactionRequest 

type CompactionRequest struct {
	CheckpointPath string
	Watermark      Version
	Records        []Record
	Base           *Checkpoint
	PeerWatermarks []Version
}

CompactionRequest describes a checkpoint and the peer acknowledgements that make truncation safe. Base is the checkpoint represented by the journal floor, if any.

type Journal 

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

Journal is an ordered, append-only lifecycle mutation log.

func (*Journal) Append 

func (j *Journal) Append(ctx context.Context, record Record) error

Append persists one strictly newer mutation and fsyncs it before returning.

func (*Journal) Bytes 

func (j *Journal) Bytes() int64

func (*Journal) Close 

func (j *Journal) Close() error

Close releases the journal file. Repeated closes are harmless.

func (*Journal) Compact 

func (j *Journal) Compact(ctx context.Context, request CompactionRequest) error

Compact durably writes the checkpoint before replacing the journal with its retained tail.

func (*Journal) Floor 

func (j *Journal) Floor() Version

func (*Journal) InstallSnapshot 

func (j *Journal) InstallSnapshot(ctx context.Context, checkpointPath string, checkpoint Checkpoint) error

InstallSnapshot durably installs a complete remote checkpoint and advances the journal floor. It is used for a peer that cannot be repaired from the retained journal tail; older local history is never retained as a suffix.

func (*Journal) LastVersion 

func (j *Journal) LastVersion() Version

func (*Journal) Len 

func (j *Journal) Len() int

func (*Journal) Records 

func (j *Journal) Records(ctx context.Context) ([]Record, error)

Records returns the retained journal tail as owned copies.

func (*Journal) RepairBatch 

func (j *Journal) RepairBatch(ctx context.Context, from Version, limits RepairLimits) (RepairBatch, error)

RepairBatch returns the next bounded journal tail after a peer watermark.

func (*Journal) Replay 

func (j *Journal) Replay(ctx context.Context, apply func(Record) error) error

Replay calls apply in durable journal order.

func (*Journal) Stats 

func (j *Journal) Stats() JournalStats

Stats returns one consistent aggregate journal snapshot.

type JournalOptions 

type JournalOptions struct {
	Limits            Limits
	MaxJournalBytes   int64
	MaxJournalEntries int
}

JournalOptions bounds the durable journal.

type JournalStats 

type JournalStats struct {
	Floor   Version
	Last    Version
	Entries int
	Bytes   int64
}

JournalStats describes the retained durable journal without returning records.

type LifecycleState 

type LifecycleState string

LifecycleState is the logical state of one application key. 

const (
	StatePresent LifecycleState = "present"
	StateDeleted LifecycleState = "deleted"
)

type Limits 

type Limits struct {
	MaxNamespaceBytes   int
	MaxLogicalKeyBytes  int
	MaxAuthorityIDBytes int
	MaxRecordBytes      int
	MaxCheckpointBytes  int
}

Limits bounds lifecycle metadata before it is persisted or exchanged.

type MembershipBook 

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

MembershipBook durably stores an operator-authored set of replica identities. A missing file is distinct from an explicitly persisted empty set so callers can fail closed when compaction has no configured safety fence.

func OpenMembershipBook 

func OpenMembershipBook(path string, options MembershipOptions) (*MembershipBook, error)

OpenMembershipBook opens a durable membership set, or an unconfigured empty book when the path has not been created yet.

func (*MembershipBook) Configured 

func (b *MembershipBook) Configured() bool

Configured reports whether an operator-authored membership file exists.

func (*MembershipBook) Progress 

func (b *MembershipBook) Progress(ctx context.Context, watermarks *WatermarkBook, target Version) (MembershipProgress, error)

Progress summarizes how many configured members have acknowledged Target and returns the minimum durable watermark without exposing member identities.

func (*MembershipBook) Replace 

func (b *MembershipBook) Replace(ctx context.Context, members []string) error

Replace durably replaces the operator-authored membership set. An empty slice is valid and creates an explicit empty fence distinct from a missing file.

func (*MembershipBook) Snapshot 

func (b *MembershipBook) Snapshot() []string

Snapshot returns an owned, stable copy of the configured identities.

func (*MembershipBook) WatermarksAt 

func (b *MembershipBook) WatermarksAt(ctx context.Context, watermarks *WatermarkBook, target Version) ([]Version, error)

WatermarksAt checks every configured member's durable acknowledgement at a compaction target and returns the ordered evidence required by Journal.Compact.

type MembershipOptions 

type MembershipOptions struct {
	MaxPeers       int
	MaxPeerIDBytes int
	MaxBytes       int
}

MembershipOptions bounds durable operator-authored membership state.

type MembershipProgress 

type MembershipProgress struct {
	Configured   bool
	Members      int
	Acknowledged int
	Minimum      Version
	Target       Version
}

MembershipProgress is a bounded aggregate of the configured membership fence. It intentionally carries no replica identity or logical data.

type Record 

type Record struct {
	Namespace   []byte         `json:"namespace"`
	LogicalKey  []byte         `json:"logical_key"`
	State       LifecycleState `json:"state"`
	BlobKey     []byte         `json:"blob_key,omitempty"`
	Version     Version        `json:"version"`
	AuthorityID string         `json:"authority_id"`
}

Record is a logical lifecycle mutation. BlobKey is present only for a present record.

func DecodeRecord 

func DecodeRecord(payload []byte, limits TransportLimits) (Record, error)

DecodeRecord validates one lifecycle record envelope without applying it.

func DecodeRecordForPeer 

func DecodeRecordForPeer(payload []byte, capabilities []string, limits TransportLimits) (Record, error)

DecodeRecordForPeer refuses lifecycle data before decoding when the peer lacks lifecycle.v1.

func (Record) Validate 

func (r Record) Validate(limits Limits) error

Validate checks bounds and the invariants required before persistence.

type RecordEnvelope 

type RecordEnvelope struct {
	Type   string `json:"type"`
	Record Record `json:"record"`
}

RecordEnvelope carries one validated lifecycle mutation over the wire.

type Recovery 

type Recovery struct {
	TruncatedTail bool
	Floor         Version
	LastVersion   Version
	Entries       int
}

Recovery describes repair performed while opening a journal.

type RepairAck 

type RepairAck struct {
	Type      string  `json:"type"`
	Watermark Version `json:"watermark"`
}

RepairAck confirms that a receiver durably applied through Watermark.

func DecodeRepairAck 

func DecodeRepairAck(payload []byte, limits RepairLimits) (RepairAck, error)

DecodeRepairAck validates one durable watermark acknowledgement.

type RepairBatch 

type RepairBatch struct {
	Type    string   `json:"type"`
	From    Version  `json:"from"`
	To      Version  `json:"to"`
	More    bool     `json:"more"`
	Records []Record `json:"records"`
}

RepairBatch is an ordered journal slice after From and through To.

func DecodeRepairBatch 

func DecodeRepairBatch(payload []byte, limits RepairLimits) (RepairBatch, error)

DecodeRepairBatch validates one ordered journal batch without applying records.

func (RepairBatch) Delivery 

func (b RepairBatch) Delivery(expected Version, limits RepairLimits) (RepairDelivery, error)

Delivery classifies a batch as ready, a harmless duplicate, or a gap.

type RepairCoordinator 

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

RepairCoordinator connects durable peer watermarks to bounded journal plans.

func NewRepairCoordinator 

func NewRepairCoordinator(journal *Journal, watermarks *WatermarkBook, limits RepairLimits) (*RepairCoordinator, error)

NewRepairCoordinator creates a planner with durable peer acknowledgement state.

func (*RepairCoordinator) Acknowledge 

func (c *RepairCoordinator) Acknowledge(ctx context.Context, peer string, version Version) error

Acknowledge persists a peer's applied watermark after ensuring it exists in the local journal. It never advances beyond the source's durable tail.

func (*RepairCoordinator) Plan 

func (c *RepairCoordinator) Plan(ctx context.Context, peer string, snapshot *Checkpoint) (RepairPlan, error)

Plan chooses the next bounded repair payload from the peer's durable watermark.

func (*RepairCoordinator) Reconcile 

func (c *RepairCoordinator) Reconcile(ctx context.Context, peer string, version Version) error

Reconcile accepts an authenticated peer's startup watermark, including a reset to zero after that peer loses its local lifecycle metadata.

type RepairDelivery 

type RepairDelivery string

RepairDelivery classifies a decoded batch relative to the receiver watermark. 

const (
	RepairDeliveryReady     RepairDelivery = "ready"
	RepairDeliveryDuplicate RepairDelivery = "duplicate"
)

type RepairFrame 

type RepairFrame struct {
	Type     string
	Batch    *RepairBatch
	Snapshot *RepairSnapshot
	Ack      *RepairAck
}

RepairFrame is the bounded lifecycle repair message union used at the wire boundary.

func DecodeRepairFrame 

func DecodeRepairFrame(payload []byte, limits RepairLimits) (RepairFrame, error)

DecodeRepairFrame validates one lifecycle repair message without applying it.

func DecodeRepairFrameForPeer 

func DecodeRepairFrameForPeer(payload []byte, capabilities []string, limits RepairLimits) (RepairFrame, error)

DecodeRepairFrameForPeer refuses lifecycle repair data before decoding when the remote peer lacks lifecycle.v1.

type RepairFramePeer 

type RepairFramePeer interface {
	AuthCapabilities() []string
	WriteFrame(payload []byte, maxPayload int) error
}

RepairFramePeer is the minimal authenticated peer surface required by a lifecycle repair session. p2p.TCPPeer implements it without a package cycle.

type RepairLimits 

type RepairLimits struct {
	MaxRecords         int
	MaxLogicalKeyBytes int
	MaxMetadataBytes   int
	MaxFrameBytes      int
	RecordLimits       Limits
}

RepairLimits bounds one ordered batch or snapshot.

type RepairPlan 

type RepairPlan struct {
	Mode    RepairPlanMode
	From    Version
	To      Version
	More    bool
	Payload []byte
}

RepairPlan is a bounded payload and its resulting watermark.

func PlanRepair 

func PlanRepair(ctx context.Context, journal *Journal, from Version, snapshot *Checkpoint, limits RepairLimits) (RepairPlan, error)

PlanRepair chooses a bounded journal batch or a checkpoint fallback.

type RepairPlanMode 

type RepairPlanMode string

RepairPlanMode identifies whether a peer receives journal tail or snapshot state. 

const (
	RepairPlanJournal  RepairPlanMode = "journal"
	RepairPlanSnapshot RepairPlanMode = "snapshot"
)

type RepairSession 

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

RepairSession plans, sends, receives, applies, and acknowledges bounded lifecycle repair frames for one authenticated peer.

func NewRepairSession 

func NewRepairSession(options RepairSessionOptions) (*RepairSession, error)

NewRepairSession constructs a cancellation-safe session. Applier is optional for a sender-only session and required when a batch or snapshot is received.

func (*RepairSession) Handle 

func (s *RepairSession) Handle(ctx context.Context, payload []byte) error

Handle decodes one incoming repair frame. Received records are applied before a watermark acknowledgement is persisted and written to the peer.

func (*RepairSession) Run 

func (s *RepairSession) Run(ctx context.Context) error

Run sends one bounded repair sequence and waits for durable acknowledgements before planning the next frame. It stops after the current journal tail; callers can start it again after a reconnect or when new records exist.

func (*RepairSession) SendNext 

func (s *RepairSession) SendNext(ctx context.Context) (RepairPlan, error)

SendNext plans and writes one bounded repair frame. The watermark advances only after the receiver sends a durable acknowledgement back through Handle.

type RepairSessionOptions 

type RepairSessionOptions struct {
	Peer           RepairFramePeer
	Coordinator    *RepairCoordinator
	Applier        *Applier
	PeerID         string
	Snapshot       *Checkpoint
	CheckpointPath string
	MaxFrameBytes  int
	ReconcilePeer  bool
	// BeforeRepair runs after startup watermark reconciliation and before the first repair frame.
	BeforeRepair func(context.Context) error
}

RepairSessionOptions configures one sender/receiver lifecycle repair session.

type RepairSnapshot 

type RepairSnapshot struct {
	Type      string   `json:"type"`
	Watermark Version  `json:"watermark"`
	Records   []Record `json:"records"`
}

RepairSnapshot is a complete logical checkpoint at Watermark.

func DecodeRepairSnapshot 

func DecodeRepairSnapshot(payload []byte, limits RepairLimits) (RepairSnapshot, error)

DecodeRepairSnapshot validates one complete checkpoint bootstrap.

type Store 

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

Store is an in-memory logical state view used by the journal and future replicas.

func NewStore 

func NewStore(limits Limits) *Store

NewStore returns an empty lifecycle state view.

func (*Store) Apply 

func (s *Store) Apply(record Record) (ApplyResult, error)

Apply applies a record according to version and same-token conflict rules.

func (*Store) Classify 

func (s *Store) Classify(record Record) (ApplyResult, error)

Classify validates a record and reports its outcome without changing state.

func (*Store) Get 

func (s *Store) Get(namespace, key []byte) (Record, bool, error)

Get returns a copy of the current record for a logical key.

func (*Store) ReplaceSnapshot 

func (s *Store) ReplaceSnapshot(records []Record) error

ReplaceSnapshot atomically replaces the logical view with a validated, duplicate-free checkpoint set. Callers persist the corresponding journal checkpoint before publishing this in-memory view.

func (*Store) Snapshot 

func (s *Store) Snapshot() []Record

Snapshot returns records in deterministic namespace/key order.

func (*Store) Stats 

func (s *Store) Stats() StoreStats

Stats returns aggregate logical-state counts without exposing key material.

func (*Store) ValidateKey 

func (s *Store) ValidateKey(namespace, key []byte) error

ValidateKey checks namespace and logical-key bounds without reading state.

type StoreStats 

type StoreStats struct {
	Records    int
	Tombstones int
}

StoreStats describes logical state without returning keys or record bodies.

type TransportLimits 

type TransportLimits struct {
	MaxPayloadBytes int
	RecordLimits    Limits
}

TransportLimits bounds a lifecycle record envelope and its inner record separately.

type Version 

type Version struct {
	Epoch    uint64 `json:"epoch"`
	Sequence uint64 `json:"sequence"`
}

Version is the total-order token assigned by a namespace authority.

func (Version) Compare 

func (v Version) Compare(other Version) int

Compare compares versions lexicographically by epoch and sequence.

func (Version) IsZero 

func (v Version) IsZero() bool

type WatermarkBook 

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

WatermarkBook durably stores monotonic lifecycle acknowledgements by peer. Every successful mutation is written through an envelope and atomic rename before becoming visible to readers.

func OpenWatermarkBook 

func OpenWatermarkBook(path string, options WatermarkOptions) (*WatermarkBook, error)

OpenWatermarkBook opens a durable acknowledgement map, or an empty one when the path has not been created yet.

func (*WatermarkBook) Acknowledge 

func (b *WatermarkBook) Acknowledge(ctx context.Context, peer string, version Version) error

Acknowledge durably advances one peer's watermark. Zero is the implicit initial watermark and does not create a peer entry.

func (*WatermarkBook) Forget 

func (b *WatermarkBook) Forget(ctx context.Context, peer string) error

Forget removes one peer acknowledgement durably. It is idempotent for an unknown peer and is intended for explicit membership removal by an operator.

func (*WatermarkBook) Reconcile 

func (b *WatermarkBook) Reconcile(ctx context.Context, peer string, version Version) error

Reconcile replaces one peer watermark with the authenticated peer's current view. Unlike Acknowledge, it permits a durable reset after peer metadata loss.

func (*WatermarkBook) Snapshot 

func (b *WatermarkBook) Snapshot() map[string]Version

Snapshot returns an owned copy of all durable peer acknowledgements.

func (*WatermarkBook) Watermark 

func (b *WatermarkBook) Watermark(peer string) Version

Watermark returns the acknowledged watermark for peer, or zero when unknown.

type WatermarkOptions 

type WatermarkOptions struct {
	MaxPeers       int
	MaxPeerIDBytes int
	MaxBytes       int
}

WatermarkOptions bounds durable per-peer acknowledgement state.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.