ledgercore

package
v0.1.3 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: AGPL-3.0 Imports: 13 Imported by: 0

Documentation

Overview

Package ledgercore provides shared primitives and infrastructure for event-sourced ledger implementations in the mivia agent.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotFound is returned when a requested run or task does not exist.
	ErrNotFound = errors.New("ledger: not found")

	// ErrConflict is returned on optimistic concurrency failures (e.g. CAS mismatch).
	ErrConflict = errors.New("ledger: conflict")

	// ErrDuplicate is returned when creating a run or task with an existing ID or key.
	ErrDuplicate = errors.New("ledger: duplicate")

	// ErrClaimHeld is returned when an exclusive run execution claim is held by another process.
	ErrClaimHeld = errors.New("ledger: run claim held by another executor")

	// ErrClaimNotHeld is returned when releasing or refreshing a claim not owned by the caller.
	ErrClaimNotHeld = errors.New("ledger: run claim not held by this caller")

	// ErrClosed is returned when an operation is attempted on a closed ledger.
	ErrClosed = errors.New("ledger: closed")

	// ErrInvalidTransition is returned when a state transition is not allowed.
	ErrInvalidTransition = errors.New("ledger: invalid state transition")

	// ErrContentNotFound is returned when content-addressed bytes are not found in the store.
	ErrContentNotFound = errors.New("ledger: content not found")
)

Common ledger error sentinels.

Functions

func LoadContent 

func LoadContent(ctx context.Context, store storage.Store, ref string) ([]byte, error)

LoadContent retrieves stored bytes under ref. It returns ErrContentNotFound if absent. When ref carries the "sha256:" prefix, the stored bytes are verified against the ref's embedded hex digest; other ref shapes are returned verbatim.

func MapStorageError 

func MapStorageError(err error) error

MapStorageError maps storage package error sentinels to ledgercore error sentinels.

func NewHolderID 

func NewHolderID() string

NewHolderID generates a random per-process identifier for run execution claims. It uses crypto/rand to guarantee uniqueness and unpredictability across processes.

func ParseClaimAcquiredAt 

func ParseClaimAcquiredAt(s string) (time.Time, error)

ParseClaimAcquiredAt parses a claim's acquired_at timestamp.

func SortEvents 

func SortEvents(events []storage.Event)

SortEvents sorts events in-place into global store order (RowID ascending, then Sequence ascending).

func SortEventsStable 

func SortEventsStable(events []storage.Event)

SortEventsStable stably sorts events in-place into global store order (RowID ascending, then Sequence ascending).

func StoreContent 

func StoreContent(ctx context.Context, store storage.Store, ref string, data []byte) error

StoreContent persists bytes under a content-addressed reference.

Types

type AppendOptions 

type AppendOptions struct {
	BoundHolder string
	Rollback    func()
	RebuildRun  func(ctx context.Context, runID string) error
	OnDuplicate func(ctx context.Context, evt storage.Event) error
}

AppendOptions controls error handling and duplicate resolution during event append.

type ClaimsTracker 

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

ClaimsTracker manages run execution claims, fenced leases, and shutdown cleanup for an event-sourced ledger repository over a storage.Store.

func NewClaimsTracker 

func NewClaimsTracker(holder string) *ClaimsTracker

NewClaimsTracker creates a new ClaimsTracker initialized with a holder ID. If holder is empty, a random holder ID is generated.

func (*ClaimsTracker) AllClaims 

func (c *ClaimsTracker) AllClaims() map[string]storage.Claim

AllClaims returns a copy of all active tracked claims.

func (*ClaimsTracker) CheckOpen 

func (c *ClaimsTracker) CheckOpen() error

CheckOpen returns ErrClosed if the tracker is closed.

func (*ClaimsTracker) ClaimRun 

func (c *ClaimsTracker) ClaimRun(ctx context.Context, store storage.Store, runID, holder string) error

ClaimRun acquires an exclusive run execution claim on store. Serialized under lock with Close() to ensure no zombie claims are left behind.

func (*ClaimsTracker) ClearRunClaim 

func (c *ClaimsTracker) ClearRunClaim(ctx context.Context, store storage.Store, runID string) error

ClearRunClaim removes a run claim (force release).

func (*ClaimsTracker) Close 

func (c *ClaimsTracker) Close(ctx context.Context, store storage.Store) error

Close marks the tracker as closed and releases all active claims held by this instance. It never closes the underlying store.

func (*ClaimsTracker) DropClaim 

func (c *ClaimsTracker) DropClaim(runID string)

DropClaim removes runID from the tracked active claims map.

func (*ClaimsTracker) GetClaim 

func (c *ClaimsTracker) GetClaim(runID string) (storage.Claim, bool)

GetClaim returns the tracked storage.Claim for runID, if present.

func (*ClaimsTracker) GetRunClaim 

func (c *ClaimsTracker) GetRunClaim(ctx context.Context, store storage.Store, runID string) (holder string, acquiredAt time.Time, ok bool, err error)

GetRunClaim reads the run's current execution claim without modifying it.

func (*ClaimsTracker) Holder 

func (c *ClaimsTracker) Holder() string

Holder returns the tracker's holder identifier.

func (*ClaimsTracker) IsClosed 

func (c *ClaimsTracker) IsClosed() bool

IsClosed reports whether the tracker has been closed.

func (*ClaimsTracker) IsRunHeld 

func (c *ClaimsTracker) IsRunHeld(ctx context.Context, store storage.Store, runID string) (bool, error)

IsRunHeld reports whether runID currently has an active claim.

func (*ClaimsTracker) IsRunTokenFenced 

func (c *ClaimsTracker) IsRunTokenFenced(ctx context.Context, store storage.Store, runID, token string) (bool, error)

IsRunTokenFenced reports whether token has been fenced out of runID.

func (*ClaimsTracker) RefreshRunClaim 

func (c *ClaimsTracker) RefreshRunClaim(ctx context.Context, store storage.Store, runID, holder string) error

RefreshRunClaim refreshes the claim's acquired_at only if already held.

func (*ClaimsTracker) ReleaseRun 

func (c *ClaimsTracker) ReleaseRun(ctx context.Context, store storage.Store, runID, holder string) error

ReleaseRun releases the claim on runID. Only the current holder may release it.

func (*ClaimsTracker) SetClaim 

func (c *ClaimsTracker) SetClaim(runID string, claim storage.Claim)

SetClaim updates the tracked storage.Claim for runID.

func (*ClaimsTracker) TakeoverExpiredRunClaim 

func (c *ClaimsTracker) TakeoverExpiredRunClaim(ctx context.Context, store storage.Store, runID, holder string, maxAge time.Duration) error

TakeoverExpiredRunClaim replaces a claim only when its age exceeds maxAge.

func (*ClaimsTracker) TakeoverRunClaim 

func (c *ClaimsTracker) TakeoverRunClaim(ctx context.Context, store storage.Store, runID, holder string) error

TakeoverRunClaim atomically replaces any existing claim on store.

type Engine 

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

Engine coordinates event-sourced ledger storage operations: claim tracking, watermark progression, sequence allocation, concurrency serialization, and durable claim-fenced appends.

func NewEngine 

func NewEngine(store storage.Store, ownsStore bool, holder string) *Engine

NewEngine initializes a new ledger Engine over store.

func (*Engine) AppendEvent 

func (e *Engine) AppendEvent(ctx context.Context, evt storage.Event, opts AppendOptions) error

AppendEvent writes evt to store with claim fencing, updating watermarks on success.

func (*Engine) CatchUp 

func (e *Engine) CatchUp(
	ctx context.Context,
	filterRun func(runID string, maxSeq int) FilterDecision,
	rebuildRun func(ctx context.Context, runID string, events []storage.Event) error,
) error

CatchUp performs incremental catch-up by probing store changes and invoking rebuildRun.

func (*Engine) CatchUpSince 

func (e *Engine) CatchUpSince(
	ctx context.Context,
	filterRun func(runID string, maxSeq int) FilterDecision,
	applyTail func(ctx context.Context, events []storage.Event) error,
) error

CatchUpSince probes store changes and fetches new events using EventsSince for all runs behind their applied watermark, passing the merged tail to applyTail.

func (*Engine) CheckDuplicatePayload 

func (e *Engine) CheckDuplicatePayload(ctx context.Context, evt storage.Event) error

CheckDuplicatePayload verifies whether a duplicate event has identical payload.

func (*Engine) CheckOpen 

func (e *Engine) CheckOpen() error

CheckOpen returns ErrClosed if the engine has been closed.

func (*Engine) ClaimRun 

func (e *Engine) ClaimRun(ctx context.Context, runID, holder string) error

ClaimRun acquires an exclusive run execution claim on store.

func (*Engine) Claims 

func (e *Engine) Claims() *ClaimsTracker

Claims returns the associated ClaimsTracker.

func (*Engine) ClearRunClaim 

func (e *Engine) ClearRunClaim(ctx context.Context, runID string) error

ClearRunClaim removes a run claim (force release).

func (*Engine) Close 

func (e *Engine) Close(ctx context.Context) error

Close closes the engine, releases active claims, and closes the store if owned.

func (*Engine) GetRunClaim 

func (e *Engine) GetRunClaim(ctx context.Context, runID string) (holder string, acquiredAt time.Time, ok bool, err error)

GetRunClaim reads the run's current execution claim without modifying it.

func (*Engine) IsClosed 

func (e *Engine) IsClosed() bool

IsClosed reports whether the engine has been closed.

func (*Engine) IsRunHeld 

func (e *Engine) IsRunHeld(ctx context.Context, runID string) (bool, error)

IsRunHeld reports whether runID currently has an active claim.

func (*Engine) IsRunTokenFenced 

func (e *Engine) IsRunTokenFenced(ctx context.Context, runID, token string) (bool, error)

IsRunTokenFenced reports whether token has been fenced out of runID.

func (*Engine) LoadContent 

func (e *Engine) LoadContent(ctx context.Context, ref string) ([]byte, error)

LoadContent retrieves stored bytes under ref.

func (*Engine) NextSequence 

func (e *Engine) NextSequence(runID string) uint64

NextSequence generates and records the next monotonic sequence for runID.

func (*Engine) Now 

func (e *Engine) Now() time.Time

Now returns the current time using the configured time source.

func (*Engine) RebaseRunSequence 

func (e *Engine) RebaseRunSequence(ctx context.Context, runID string) error

RebaseRunSequence reads existing events for runID from store and updates watermarks.

func (*Engine) RefreshRunClaim 

func (e *Engine) RefreshRunClaim(ctx context.Context, runID, holder string) error

RefreshRunClaim refreshes the claim's acquired_at only if already held.

func (*Engine) ReleaseRun 

func (e *Engine) ReleaseRun(ctx context.Context, runID, holder string) error

ReleaseRun releases the claim on runID. Only the current holder may release it.

func (*Engine) RunLock 

func (e *Engine) RunLock(runID string) *sync.Mutex

RunLock returns the mutex that serializes operations on runID.

func (*Engine) SetTimeSource 

func (e *Engine) SetTimeSource(now func() time.Time)

SetTimeSource configures a custom clock for deterministic testing.

func (*Engine) Store 

func (e *Engine) Store() storage.Store

Store returns the underlying storage.Store.

func (*Engine) StoreContent 

func (e *Engine) StoreContent(ctx context.Context, ref string, data []byte) error

StoreContent persists bytes under a content-addressed reference.

func (*Engine) TakeoverExpiredRunClaim 

func (e *Engine) TakeoverExpiredRunClaim(ctx context.Context, runID, holder string, maxAge time.Duration) error

TakeoverExpiredRunClaim replaces a claim only when its age exceeds maxAge.

func (*Engine) TakeoverRunClaim 

func (e *Engine) TakeoverRunClaim(ctx context.Context, runID, holder string) error

TakeoverRunClaim atomically replaces any existing claim on store.

func (*Engine) Watermarks 

func (e *Engine) Watermarks() *WatermarkTracker

Watermarks returns the associated WatermarkTracker.

type FilterDecision 

type FilterDecision int

FilterDecision indicates how CatchUp should handle a behind run. 

const (
	// FilterApply indicates the run should be locked, fetched, and rebuilt.
	FilterApply FilterDecision = iota
	// FilterSkip indicates the run should be skipped without updating its watermark.
	FilterSkip
	// FilterAdvanceOnly indicates the watermark should advance without reading events.
	FilterAdvanceOnly
)

type RunPolicy 

type RunPolicy struct {
	NoRetry         bool `json:"no_retry"`
	FailInterrupted bool `json:"fail_interrupted"`
	// Retry fields are the immutable scheduler policy captured at admission.
	// They are work policy, not caller authority.
	RetryMaxRetries     int           `json:"retry_max_retries"`
	RetryBaseBackoff    time.Duration `json:"retry_base_backoff"`
	RetryMaxBackoff     time.Duration `json:"retry_max_backoff"`
	RetryBackoffFactor  float64       `json:"retry_backoff_factor"`
	RetryJitterFraction float64       `json:"retry_jitter_fraction"`
}

RunPolicy describes fixed recovery behaviour for one admitted run.

type WatermarkTracker 

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

WatermarkTracker tracks applied sequence watermarks, allocated sequence numbers, and probe cursors for an event-sourced repository over a storage.Store.

func NewWatermarkTracker 

func NewWatermarkTracker() *WatermarkTracker

NewWatermarkTracker initializes a new WatermarkTracker.

func (*WatermarkTracker) AdvanceCursor 

func (w *WatermarkTracker) AdvanceCursor(cursor uint64)

AdvanceCursor moves the store probe cursor forward. It never rewinds.

func (*WatermarkTracker) Allocated 

func (w *WatermarkTracker) Allocated(runID string) uint64

Allocated returns the highest sequence allocated for runID.

func (*WatermarkTracker) Applied 

func (w *WatermarkTracker) Applied(runID string) uint64

Applied returns the highest store sequence folded into the projection for runID.

func (*WatermarkTracker) CheckBehind 

func (w *WatermarkTracker) CheckBehind(maxSequences map[string]int) []string

CheckBehind compares the store's maxSequences against applied watermarks and returns a sorted list of run IDs that have unapplied changes.

func (*WatermarkTracker) Cursor 

func (w *WatermarkTracker) Cursor() uint64

Cursor returns the highest store append position probed so far.

func (*WatermarkTracker) DeleteRun 

func (w *WatermarkTracker) DeleteRun(runID string)

DeleteRun clears all sequence and watermark tracking for runID.

func (*WatermarkTracker) NextSequence 

func (w *WatermarkTracker) NextSequence(runID string) uint64

NextSequence computes and reserves the next sequence number for runID.

func (*WatermarkTracker) RebaseRunSequence 

func (w *WatermarkTracker) RebaseRunSequence(ctx context.Context, store storage.Store, runID string) error

RebaseRunSequence reads existing events for runID from store and ensures applied and allocated are at least the highest sequence already in the store.

func (*WatermarkTracker) SetAllocated 

func (w *WatermarkTracker) SetAllocated(runID string, seq uint64)

SetAllocated sets the allocated sequence for runID.

func (*WatermarkTracker) SetApplied 

func (w *WatermarkTracker) SetApplied(runID string, seq uint64)

SetApplied sets the applied watermark for runID.

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.