Documentation
ยถ
Overview ยถ
Package sop defines the core interfaces, types, and helpers used across the SOP codebase. It provides transactions, store options and metadata, key/handle abstractions, and shared error codes. Concrete backends live in subpackages such as fs (filesystem), cassandra, and redis, while higher-level features include B-Trees and streaming data helpers. It is designed to be extensible and modular, allowing for various storage backends to be implemented while sharing a common interface. This package is intended for internal use within the SOP project and is not meant for external use. It is a foundational package that other components build upon. It is not intended to be used directly by end-users, but rather serves as a base for more specific implementations and utilities in the SOP ecosystem. It is a foundational package that other components build upon. It is not intended to be used directly by end-users, but rather serves as a base for more specific implementations and utilities in the SOP ecosystem.
See `infs.package` for a concrete implementation of a File System-based store with built-in Redis caching.
Package sop contains SOP integration code for Redis, Cassandra & Kafka (in_red_c).
Index ยถ
- Constants
- Variables
- func Authorize(ctx context.Context, access ResourceAccess, action Action) bool
- func CanPerformAction(ctx context.Context, resourceName string, access ResourceAccess, action Action) bool
- func CheckPolicy(ctx context.Context, resourceName string, access ResourceAccess, action Action) error
- func ConfigureLogging()
- func ContextWithAuth(ctx context.Context, auth AuthContext) context.Context
- func EnforcePolicy(ctx context.Context, resourceName string, access ResourceAccess, action Action) error
- func FlattenForSchema(key any, value any) map[string]any
- func FormatRegistryTable(name string) string
- func FormatSchema(schema map[string]string) string
- func GetAllBlueprints() map[string]AssetBlueprint
- func HandleLockAcquisitionFailure(ctx context.Context, err error, rollbackFunc func(context.Context, UUID) error, ...) error
- func InferSchema(item map[string]any) map[string]string
- func InferSchemaType(v any) string
- func InferType(v any) (string, bool)
- func IsFailoverQualifiedIOError(err error) bool
- func IsSystemReadOnly(resourceName string) bool
- func RandomSleep(ctx context.Context)
- func RandomSleepWithUnit(ctx context.Context, unit time.Duration)
- func RegisterAssetRBAC(blueprint AssetBlueprint)
- func RegisterL2CacheFactory(ct L2CacheType, f L2CacheFactory)
- func Retry(ctx context.Context, task func(ctx context.Context) error, ...) error
- func SetDefaultCacheConfig(cacheDuration StoreCacheConfig)
- func SetJitterRNG(r *rand.Rand)
- func SetLogLevel(level slog.Level)
- func ShouldRetry(err error) bool
- func Sleep(ctx context.Context, sleepTime time.Duration)
- func TimedOut(ctx context.Context, name string, startTime time.Time, maxTime time.Duration) error
- type Action
- type AssetBlueprint
- type AuthContext
- type BlobStore
- type BlobsPayload
- type BundledResponse
- type CloseableCache
- type ContextRBACMap
- type DatabaseOptions
- func (do DatabaseOptions) CopyTo(transOptions *TransactionOptions)
- func (do DatabaseOptions) GetDatabaseType() DatabaseType
- func (do DatabaseOptions) IsCassandraHybrid() bool
- func (do DatabaseOptions) IsEmpty() bool
- func (do DatabaseOptions) IsReplicated() bool
- func (do *DatabaseOptions) SetDatabaseType(t DatabaseType)
- type DatabaseType
- type EndpointContext
- type EntitlementContext
- type ErasureCodingConfig
- type ErrTimeout
- type Error
- type ErrorCode
- type Handle
- func (h *Handle) AllocateID() UUID
- func (h *Handle) ClearInactiveID()
- func (h *Handle) FlipActiveID()
- func (h Handle) GetActiveID() UUID
- func (h Handle) GetInActiveID() UUID
- func (h *Handle) HasID(id UUID) bool
- func (h Handle) IsAandBinUse() bool
- func (x *Handle) IsEmpty() bool
- func (x *Handle) IsEqual(y *Handle) bool
- func (h *Handle) IsExpiredInactive() bool
- type KeyValuePair
- type KeyValueStore
- type KeyValueStoreItemActionResponse
- type KeyValueStoreResponse
- type L2Cache
- type L2CacheFactory
- type L2CacheType
- type LockKey
- type Locker
- type ManageStore
- type RedisCacheConfig
- type Registry
- type RegistryPayload
- type Relation
- type ResourceAccess
- type SchemaInferenceResult
- type SinglePhaseTransaction
- func (t *SinglePhaseTransaction) AddPhasedTransaction(otherTransaction ...TwoPhaseCommitTransaction)
- func (t *SinglePhaseTransaction) Begin(ctx context.Context) error
- func (t *SinglePhaseTransaction) Close() error
- func (t *SinglePhaseTransaction) Commit(ctx context.Context) error
- func (t *SinglePhaseTransaction) CommitMaxDuration() time.Duration
- func (t *SinglePhaseTransaction) GetID() UUID
- func (t *SinglePhaseTransaction) GetMode() TransactionMode
- func (t *SinglePhaseTransaction) GetPhasedTransaction() TwoPhaseCommitTransaction
- func (t *SinglePhaseTransaction) GetStores(ctx context.Context) ([]string, error)
- func (t *SinglePhaseTransaction) HasBegun() bool
- func (t *SinglePhaseTransaction) OnCommit(callback func(ctx context.Context) error)
- func (t *SinglePhaseTransaction) Rollback(ctx context.Context) error
- type StoreCacheConfig
- type StoreInfo
- func (si *StoreInfo) DeleteCustomData(key string) bool
- func (si *StoreInfo) GetCustomData(key string) (any, bool)
- func (si *StoreInfo) GetCustomDataMap() map[string]any
- func (s StoreInfo) IsCompatible(b StoreInfo) bool
- func (s StoreInfo) IsEmpty() bool
- func (si *StoreInfo) SetCustomData(key string, value any)
- func (si *StoreInfo) SetCustomDataMap(data map[string]any)
- func (si *StoreInfo) ValueDataSize() ValueDataSize
- type StoreOptions
- type StoreRepository
- type TaskRunner
- type Transaction
- type TransactionLog
- type TransactionMode
- type TransactionOptions
- type TransactionPriorityLog
- type Tuple
- type TwoPhaseCommitTransaction
- type UICapability
- type UUID
- type ValueDataSize
- type Visibility
Constants ยถ
const ( // Unknown represents an unspecified error condition. Unknown ErrorCode = iota // LockAcquisitionFailure indicates failure to acquire a required lock. LockAcquisitionFailure // FileIOError represents file I/O related errors, e.g. encountered by BlobStore (w/ & w/o EC). // This should not generate Failover event because BlobStore errors are either handled internally for no EC // or by EC replication feature. FileIOError // FailoverQualifiedError marks an error that qualifies the operation for failover handling. FailoverQualifiedError = 77 + iota // FileIOErrorFailoverQualified represents file I/O related errors. FileIOErrorFailoverQualified // RestoreRegistryFileSectorFailure indicates a failure while restoring a registry file sector. RestoreRegistryFileSectorFailure )
const ( RoleAdmin = "Admin" RoleUser = "User" RoleGuest = "Guest" )
const ContextPriorityLogIgnoreAge contextKey = "plg_ignore_age"
ContextPriorityLogIgnoreAge signals priority log GetBatch to ignore age filter when true.
const (
// HandleSizeInBytes is the size, in bytes, of a Handle structure when encoded.
HandleSizeInBytes = 62
)
const MaxSlotLength = 20000
MaxSlotLength is the maximum number of items a node can accommodate. Enforced both when a StoreInfo is first created (NewStoreInfo) and again at the point a node is allocated (btree.getSlotLength), since a StoreInfo loaded from persisted storage doesn't go through NewStoreInfo and could otherwise carry an unbounded or corrupted value straight into a make().
Variables ยถ
var ( ErrSystemReadOnly = errors.New("system knowledge bases are read-only") ErrQuotaExceeded = errors.New("quota exceeded") )
var Now = time.Now
Now returns the current time. It is a var to allow tests to override time.Now for determinism.
var RetryStartDuration = 1 * time.Second
RetryStartDuration is the initial duration for the Fibonacci backoff. It is exported to allow tests to reduce the wait time.
var Version = strings.TrimSpace(versionFile)
Version is the current version of the SOP library/application.
Functions ยถ
func CanPerformAction ยถ
func CanPerformAction(ctx context.Context, resourceName string, access ResourceAccess, action Action) bool
CanPerformAction checks if the current context has permission to perform the action on the resource. It returns a boolean, making it ideal for UI visibility toggles like IsReadOnly.
func CheckPolicy ยถ
func CheckPolicy(ctx context.Context, resourceName string, access ResourceAccess, action Action) error
CheckPolicy evaluates the three-layer RBAC model and returns an error if access is denied. It is useful when you need to know exactly *why* access was denied.
func ConfigureLogging ยถ
func ConfigureLogging()
ConfigureLogging sets up the global default logger with a TextHandler and configures the log level based on the SOP_LOG_LEVEL environment variable. It defaults to Info level if not specified.
This function should be called by the application at startup if it wants to use the default SOP logging configuration.
func ContextWithAuth ยถ
func ContextWithAuth(ctx context.Context, auth AuthContext) context.Context
func EnforcePolicy ยถ
func EnforcePolicy(ctx context.Context, resourceName string, access ResourceAccess, action Action) error
EnforcePolicy checks the policy and returns an error if the action is not allowed. Use CanPerformAction for a boolean result (e.g. for adjusting UI states).
func FlattenForSchema ยถ
FlattenForSchema converts key and value of any type into a flat map[string]any suitable for schema inference. Uses JSON marshaling to handle structs. NOTE: InferSchemaFromTypes is preferred when type information is available.
func FormatRegistryTable ยถ
FormatRegistryTable formats a store name into a registry table name by adding an _r suffix.
func FormatSchema ยถ
FormatSchema formats a schema map as a sorted string like "{field: type, ...}".
func GetAllBlueprints ยถ
func GetAllBlueprints() map[string]AssetBlueprint
GetAllBlueprints returns a cloned map of the system-wide blueprints
func HandleLockAcquisitionFailure ยถ
func HandleLockAcquisitionFailure(ctx context.Context, err error, rollbackFunc func(context.Context, UUID) error, unlockFunc func(context.Context, []*LockKey) error) error
HandleLockAcquisitionFailure checks if the error is a LockAcquisitionFailure and if so, attempts to rollback the blocking transaction using the provided rollbackFunc. If rollback succeeds, it takes over the lock and releases it using unlockFunc. Returns nil if the failure was handled (lock taken over and released), otherwise returns the original error.
func InferSchema ยถ
InferSchema inspects a map and returns a simplified type definition (e.g. {"id": "uuid", "age": "number"}).
func InferSchemaType ยถ
InferSchemaType returns a string representation of the type of a value for schema inference.
func InferType ยถ
InferType returns the simplified type name (e.g. "string", "int", "uuid") and whether it's an array. This is used for UI display and loose type checking.
func IsFailoverQualifiedIOError ยถ
IsFailoverQualifiedIOError reports whether an error indicates the active drive/filesystem is unhealthy in a way that warrants immediate failover to the passive drive.
This is distinct from ShouldRetry: retryable/transient errors should be retried first, while this function targets permanent/media/FS/device conditions where staying on the current drive is counterproductive.
Notes:
- It includes common POSIX errno values available on macOS/Linux/BSD.
- For Linux-specific errno that may not exist on other platforms as named constants, numeric values are used via syscall.Errno to remain portable (they will simply never match on platforms that don't produce them).
- EFBIG is intentionally excluded per current SOP usage (registry/store repo use small files).
func IsSystemReadOnly ยถ
IsSystemReadOnly returns true if the specified resource is a core system resource that must remain read-only to prevent destructive actions by any user.
func RandomSleep ยถ
RandomSleep sleeps for a random duration between 20ms and 80ms to stagger retries.
func RandomSleepWithUnit ยถ
RandomSleepWithUnit sleeps for a random multiple (1..4) of the provided unit duration. Useful to jitter conflicting transactions and reduce contention.
func RegisterAssetRBAC ยถ
func RegisterAssetRBAC(blueprint AssetBlueprint)
RegisterAssetRBAC ensures that new Assets declare their UI footprint and execution logic cleanly
func RegisterL2CacheFactory ยถ
func RegisterL2CacheFactory(ct L2CacheType, f L2CacheFactory)
RegisterL2CacheFactory registers a cache factory for a given type.
func Retry ยถ
func Retry(ctx context.Context, task func(ctx context.Context) error, gaveUpTask func(ctx context.Context)) error
Retry executes task with Fibonacci backoff up to 5 retries. If retries are exhausted, gaveUpTask is invoked (when not nil) and the final error is returned.
func SetDefaultCacheConfig ยถ
func SetDefaultCacheConfig(cacheDuration StoreCacheConfig)
SetDefaultCacheConfig assigns the global default cache configuration used when a store does not override it.
func SetJitterRNG ยถ
SetJitterRNG overrides the RNG used for sleep jitter. Useful for deterministic tests.
func SetLogLevel ยถ
SetLogLevel sets the logging level for the logger configured by ConfigureLogging.
func ShouldRetry ยถ
ShouldRetry reports whether the error is retryable (non-nil and not a known permanent failure).
Types ยถ
type AssetBlueprint ยถ
type AssetBlueprint struct {
AssetType string // e.g., "space", "store", "agent"
Description string // Human-readable context
Endpoints []string // UI API endpoints related to this asset
Actions []Action // Capabilities: Read, Write, Delete, Execute
// Evaluator executes the actual permission check for a specific asset instance or context
Evaluator func(ctx context.Context, entitlementCtx EntitlementContext, action Action) bool
}
AssetBlueprint defines the RBAC footprint for a specific module or asset type.
func GetAssetBlueprint ยถ
func GetAssetBlueprint(assetType string) (AssetBlueprint, bool)
GetAssetBlueprint looks up the evaluator instructions by alias
type AuthContext ยถ
func GetAuthFromContext ยถ
func GetAuthFromContext(ctx context.Context) AuthContext
type BlobStore ยถ
type BlobStore interface {
// GetOne fetches a blob by ID from a blob table.
GetOne(ctx context.Context, blobTable string, blobID UUID) ([]byte, error)
// Add inserts blobs.
Add(ctx context.Context, blobs []BlobsPayload[KeyValuePair[UUID, []byte]]) error
// Update modifies existing blobs.
Update(ctx context.Context, blobs []BlobsPayload[KeyValuePair[UUID, []byte]]) error
// Remove deletes blobs by ID.
Remove(ctx context.Context, blobsIDs []BlobsPayload[UUID]) error
}
BlobStore defines CRUD operations for binary blobs that are too large for typical databases and are stored in external systems (e.g., S3, filesystem, Cassandra partitions).
type BlobsPayload ยถ
type BlobsPayload[T UUID | KeyValuePair[UUID, []byte]] struct { // BlobTable is the blob store table name (or base filesystem path). BlobTable string // Blobs holds either IDs (for deletes) or ID+data pairs (for upserts). Blobs []T }
BlobsPayload is a request/response envelope for blob operations.
type BundledResponse ยถ
type BundledResponse struct {
Data interface{} `json:"data"`
RBAC ContextRBACMap `json:"rbac,omitempty"`
ItemRBAC map[string]ContextRBACMap `json:"item_rbac,omitempty"`
}
BundledResponse represents the standard JSON payload structure containing both the domain data and the paired RBAC map.
type CloseableCache ยถ
CloseableCache is a Cache that also implements io.Closer for explicit lifecycle control.
type ContextRBACMap ยถ
type ContextRBACMap map[UICapability]bool
ContextRBACMap represents the UI-consumable map format representing capabilities for a given context: (Capability -> bool)
func ResolveRBACMap ยถ
func ResolveRBACMap(ctx context.Context, assetType string, entitlementCtx EntitlementContext, getLocalAccess func() ResourceAccess) ContextRBACMap
ResolveRBACMap evaluates the dynamic AssetBlueprint for a functional context (e.g., current space/store), delegating to the core evaluator.
type DatabaseOptions ยถ
type DatabaseOptions struct {
// StoresFolders specifies the folders for replication.
// If more than one folder, i.e. - one for Active drive/folder,
// & another for Passive drive/folder, Registry replication is enabled.
StoresFolders []string `json:"stores_folders,omitempty"`
// ErasureConfig specifies the erasure coding configuration for Blob store replication.
ErasureConfig map[string]ErasureCodingConfig `json:"erasure_config,omitempty"`
// Keyspace to be used for the transaction (Cassandra).
Keyspace string `json:"keyspace,omitempty"`
// CacheType specifies the type of cache to use (e.g. InMemory, Redis).
CacheType L2CacheType `json:"cache_type"`
// RedisConfig specifies the Redis configuration when CacheType is Redis.
RedisConfig *RedisCacheConfig `json:"redis_config,omitempty"`
// Registry hash modulo value used for hashing.
RegistryHashModValue int `json:"registry_hash_mod,omitempty"`
// Type specifies the database type (Standalone or Clustered).
// This is a convenience field that sets the default CacheType.
Type DatabaseType `json:"type"`
// EnableObfuscation specifies if this database should be obfuscated when accessed by AI tools.
// This is a runtime-only field and is not persisted to JSON.
EnableObfuscation bool `json:"-"`
}
DatabaseOptions holds the configuration for the database.
func (DatabaseOptions) CopyTo ยถ
func (do DatabaseOptions) CopyTo(transOptions *TransactionOptions)
Copy Database Options to Transaction Options.
func (DatabaseOptions) GetDatabaseType ยถ
func (do DatabaseOptions) GetDatabaseType() DatabaseType
func (DatabaseOptions) IsCassandraHybrid ยถ
func (do DatabaseOptions) IsCassandraHybrid() bool
func (DatabaseOptions) IsEmpty ยถ
func (do DatabaseOptions) IsEmpty() bool
IsEmpty returns true if database config is considered empty, i.e. - missing folder is primary. A Database should always have folder(s) where Registry data files are/will be stored.
func (DatabaseOptions) IsReplicated ยถ
func (do DatabaseOptions) IsReplicated() bool
func (*DatabaseOptions) SetDatabaseType ยถ
func (do *DatabaseOptions) SetDatabaseType(t DatabaseType)
type DatabaseType ยถ
type DatabaseType int
const ( // Standalone mode uses an in-memory cache for coordination (locks, etc.). // It is appropriate for standalone or embedded applications running in a single process. Standalone DatabaseType = iota // Clustered mode uses Redis for coordination (locks, etc.). // It allows hosting multiple application instances across a network, properly orchestrated by SOP. Clustered )
type EndpointContext ยถ
type EndpointContext string
EndpointContext represents an API endpoint representing a grouping of assets.
const ( EndpointSpacesList EndpointContext = "/api/spaces" EndpointStoresList EndpointContext = "/api/stores" EndpointItemsList EndpointContext = "/api/spaces/items" )
type EntitlementContext ยถ
type EntitlementContext struct {
AssetID string
Database string
IsSystemDB bool
UserRole string
UserID string
}
EntitlementContext holds the request scope parameters necessary for granular RBAC evaluation.
type ErasureCodingConfig ยถ
type ErasureCodingConfig struct {
// DataShardsCount is the number of data shards.
DataShardsCount int `json:"data_shards_count"`
// ParityShardsCount is the number of parity shards.
ParityShardsCount int `json:"parity_shards_count"`
// BaseFolderPathsAcrossDrives lists the drive base paths where data and parity shard files are stored.
BaseFolderPathsAcrossDrives []string `json:"base_folder_paths_across_drives"`
// RepairCorruptedShards indicates whether to attempt automatic repair when corrupted shards are detected.
// Auto-repair can be expensive; applications can disable it to prioritize throughput and handle drive
// failures via external workflows.
RepairCorruptedShards bool `json:"repair_corrupted_shards"`
}
ErasureCodingConfig defines per-blob-table erasure coding settings, including shard counts, storage locations, and optional automatic shard repair.
type ErrTimeout ยถ
type ErrTimeout struct {
// Name is a short label for the operation (e.g., "transaction", "lockFileBlockRegion").
Name string
// MaxTime is the maximum duration allowed for the operation when applicable.
MaxTime time.Duration
// Cause is the underlying timeout/cancellation cause, typically a context error.
Cause error
}
ErrTimeout is returned when an operation exceeds its allowed time budget.
Semantics:
- If a context cancellation or deadline triggered the timeout, Cause carries the original context error (context.Canceled or context.DeadlineExceeded). Unwrap() returns that Cause so errors.Is(err, context.DeadlineExceeded) works.
- If the operation-specific maximum duration triggered the timeout, Cause may be nil; MaxTime contains the configured bound for the operation.
This enables callers to branch on timeouts consistently while preserving the original context semantics when applicable.
func (ErrTimeout) Error ยถ
func (e ErrTimeout) Error() string
func (ErrTimeout) Unwrap ยถ
func (e ErrTimeout) Unwrap() error
Unwrap exposes the underlying cause (e.g., context.DeadlineExceeded) for errors.Is/As.
type Error ยถ
Error is a SOP-specific error carrying a code, the wrapped error and optional user data.
type ErrorCode ยถ
type ErrorCode int
ErrorCode enumerates SOP error categories used across packages.
type Handle ยถ
type Handle struct {
// LogicalID is the stable identifier of the entity.
LogicalID UUID
// PhysicalIDA is one of the two physical IDs supported.
PhysicalIDA UUID
// PhysicalIDB is the second physical ID supported.
PhysicalIDB UUID
// IsActiveIDB indicates whether PhysicalIDB is currently the active ID.
IsActiveIDB bool
// Version is the current state version (active ID, final deleted state).
Version int32
// WorkInProgressTimestamp stores the millisecond timestamp of the inactive ID (or non-final deleted state).
WorkInProgressTimestamp int64
// IsDeleted marks a logical delete.
IsDeleted bool
}
Handle holds a logical ID and its two physical IDs (A and B) used to implement ACID-safe swaps. SOP uses Handle to quickly switch between node versions and to support logical deletes.
func NewHandle ยถ
NewHandle creates a new Handle with the provided logical ID. PhysicalIDA is initialized to the same value.
func (*Handle) AllocateID ยถ
AllocateID generates a new UUID and assigns it to the available physical slot. If both A and B are already in use, NilUUID is returned.
func (*Handle) ClearInactiveID ยถ
func (h *Handle) ClearInactiveID()
ClearInactiveID resets the inactive physical ID to NilUUID and clears the WIP timestamp.
func (*Handle) FlipActiveID ยถ
func (h *Handle) FlipActiveID()
FlipActiveID switches the active physical ID from A to B or B to A.
func (Handle) GetActiveID ยถ
GetActiveID returns the currently active UUID (either PhysicalIDA or PhysicalIDB).
func (Handle) GetInActiveID ยถ
GetInActiveID returns the currently inactive physical UUID.
func (Handle) IsAandBinUse ยถ
IsAandBinUse reports whether both physical IDs A and B are populated.
func (*Handle) IsEmpty ยถ
IsEmpty reports whether all Handle fields are zero values (no IDs, not deleted, zero version and timestamps).
func (*Handle) IsEqual ยถ
IsEqual reports whether two Handle instances are equal ignoring the Version field.
func (*Handle) IsExpiredInactive ยถ
IsExpiredInactive reports whether the inactive ID has expired based on a fixed window.
type KeyValuePair ยถ
type KeyValuePair[TK any, TV any | []byte] struct { // Key is the key part in the pair. Key TK // Value is the value part in the pair. Value TV }
KeyValuePair represents a pair of key and value, commonly used for blob operations where the key may differ from the blob ID.
type KeyValueStore ยถ
type KeyValueStore[TK any, TV any] interface { // Fetch retrieves entries by keys from the remote storage subsystem. Fetch(context.Context, string, []TK) KeyValueStoreResponse[KeyValuePair[TK, TV]] // FetchLargeObject retrieves a single large entry by key. FetchLargeObject(context.Context, string, TK) (TV, error) // Add inserts entries. Add(context.Context, string, []KeyValuePair[TK, TV]) KeyValueStoreResponse[KeyValuePair[TK, TV]] // Update modifies existing entries. Update(context.Context, string, []KeyValuePair[TK, TV]) KeyValueStoreResponse[KeyValuePair[TK, TV]] // Remove deletes entries by keys. Remove(context.Context, string, []TK) KeyValueStoreResponse[TK] }
KeyValueStore defines CRUD operations for a generic key-value backend with optional partial success semantics.
type KeyValueStoreItemActionResponse ยถ
KeyValueStoreItemActionResponse is the per-item response including payload and error for a CRUD action.
type KeyValueStoreResponse ยถ
type KeyValueStoreResponse[T any] struct { // Details contains per-item action results. Details []KeyValueStoreItemActionResponse[T] // Error is a summary error if at least one action failed. Error error }
KeyValueStoreResponse aggregates per-item results and an optional summary error.
type L2Cache ยถ
type L2Cache interface {
// Inherit Locking interface.
Locker
// Implement to return the CacheType.
GetType() L2CacheType
// Set upserts a value under a key, and specifies when it will expire or disappear from L2 cache.
Set(ctx context.Context, key string, value string, expiration time.Duration) error
// Get returns: found(bool), value(string), err(error from backend).
Get(ctx context.Context, key string) (bool, string, error)
// GetEx returns found(bool), value(string), err using TTL/sliding expiration semantics.
GetEx(ctx context.Context, key string, expiration time.Duration) (bool, string, error)
// IsRestarted reports whether the cache backend (e.g., Redis) has restarted since the last check.
// Implementations should return true once per backend restart event per-process and false otherwise.
IsRestarted(ctx context.Context) bool
// SetStruct upserts a struct value under a key.
SetStruct(ctx context.Context, key string, value interface{}, expiration time.Duration) error
// SetStructs upserts multiple struct values under the given keys in a single round trip (pipelined).
SetStructs(ctx context.Context, keys []string, values []interface{}, expiration time.Duration) error
// GetStruct fetches a struct value; first return indicates success (false for not found or error).
GetStruct(ctx context.Context, key string, target interface{}) (bool, error)
// GetStructEx fetches a struct value with TTL/sliding expiration semantics.
GetStructEx(ctx context.Context, key string, target interface{}, expiration time.Duration) (bool, error)
// GetStructs fetches multiple struct values with optional TTL/sliding expiration semantics.
// If expiration > 0, it behaves like GetStructEx for each key (pipelined).
// If expiration <= 0, it behaves like GetStruct but batched (e.g. MGET).
GetStructs(ctx context.Context, keys []string, targets []interface{}, expiration time.Duration) ([]bool, error)
// Delete removes objects by keys; returns whether all keys were deleted.
Delete(ctx context.Context, keys []string) (bool, error)
// Ping checks connectivity to the cache backend.
Ping(ctx context.Context) error
// Clear purges the entire cache database.
Clear(ctx context.Context) error
}
L2Cache abstracts an out-of-process cache (e.g., Redis) and its locking facilities.
func GetL2Cache ยถ
func GetL2Cache(options TransactionOptions) L2Cache
GetL2Cache gets the cache (client) for the specified type. It returns nil if no factory is registered for that type.
type L2CacheFactory ยถ
type L2CacheFactory func(TransactionOptions) L2Cache
L2CacheFactory defines the function signature for creating a cache client.
type L2CacheType ยถ
type L2CacheType int
L2CacheType defines the type of cache to use.
const ( // Default represents no (L2) caching. NoCache L2CacheType = iota // InMemory represents an in-memory cache. InMemory // Redis represents a Redis cache. Redis )
type Locker ยถ
type Locker interface {
// FormatLockKey creates a lock key name from an arbitrary string.
FormatLockKey(k string) string
// CreateLockKeys builds LockKey objects from a set of key names.
CreateLockKeys(keys []string) []*LockKey
// CreateLockKeysForIDs builds LockKey objects for ID tuples (e.g., Transaction ID scoped locks).
CreateLockKeysForIDs(keys []Tuple[string, UUID]) []*LockKey
// IsLockedTTL reports whether all keys are locked and refreshes TTL with the provided duration.
IsLockedTTL(ctx context.Context, duration time.Duration, lockKeys []*LockKey) (bool, error)
// Lock attempts to lock all keys; returns success, lock owner UUID, and any error encountered.
Lock(ctx context.Context, duration time.Duration, lockKeys []*LockKey) (bool, UUID, error)
// DualLock attempts to lock all keys; returns success, lock owner UUID, and any error encountered.
// It calls Lock then IsLocked to ensure the lock is acquired and persisted.
DualLock(ctx context.Context, duration time.Duration, lockKeys []*LockKey) (bool, UUID, error)
// IsLocked reports whether all keys are currently locked.
IsLocked(ctx context.Context, lockKeys []*LockKey) (bool, error)
// IsLockedByOthers reports whether the keys are locked by other processes.
IsLockedByOthers(ctx context.Context, lockKeyNames []string) (bool, error)
// IsLockedByOthersTTL reports whether the keys are locked by other processes and refreshes TTL with the provided duration.
IsLockedByOthersTTL(ctx context.Context, lockKeyNames []string, duration time.Duration) (bool, error)
// Unlock releases a set of keys.
Unlock(ctx context.Context, lockKeys []*LockKey) error
}
Locker defines lightweight lock management facade.
type ManageStore ยถ
type ManageStore interface {
// CreateStore creates the store(s) container (e.g., a filesystem folder).
CreateStore(context.Context, string) error
// RemoveStore removes the store(s) container (e.g., a filesystem folder).
RemoveStore(context.Context, string) error
}
ManageStore declares lifecycle operations for creating and removing store containers (e.g., folders).
type RedisCacheConfig ยถ
type RedisCacheConfig struct {
// Address is the host:port of the Redis server/cluster.
Address string `json:"address"`
// Password is the password used to authenticate.
Password string `json:"password"`
// DB is the database index to select.
DB int `json:"db"`
// URL is the connection string (e.g. redis://user:pass@host:port/db).
// If provided, it overrides Address, Password, and DB.
URL string `json:"url,omitempty"`
// DialTimeout specifies the timeout for connecting to Redis.
DialTimeout time.Duration `json:"dial_timeout,omitempty"`
// ReadTimeout specifies the timeout for reading from Redis.
ReadTimeout time.Duration `json:"read_timeout,omitempty"`
// WriteTimeout specifies the timeout for writing to Redis.
WriteTimeout time.Duration `json:"write_timeout,omitempty"`
// MaxRetries is the maximum number of retries before giving up on Redis connection.
MaxRetries int `json:"max_retries,omitempty"`
}
RedisCacheConfig holds configuration for connecting to a Redis server or cluster.
type Registry ยถ
type Registry interface {
// Get fetches Handles (given logical IDs) from registry table(s).
Get(context.Context, []RegistryPayload[UUID]) ([]RegistryPayload[Handle], error)
// Add inserts Handles into registry table(s).
Add(context.Context, []RegistryPayload[Handle]) error
// Update modifies Handles across registry table(s) and acquires cache locks for each Handle.
Update(ctx context.Context, handles []RegistryPayload[Handle]) error
// UpdateNoLocks updates Handles in an active transaction where locks were pre-acquired by the transaction manager.
UpdateNoLocks(ctx context.Context, allOrNothing bool, storesHandles []RegistryPayload[Handle]) error
// Remove deletes Handles (given logical IDs) from registry table(s).
Remove(context.Context, []RegistryPayload[UUID]) error
// Replicate performs post-commit replication of blobs/data to passive targets.
Replicate(ctx context.Context, newRootNodesHandles, addedNodesHandles, updatedNodesHandles, removedNodesHandles []RegistryPayload[Handle]) error
}
Registry provides CRUD and replication operations for virtual ID management that back SOP's ACID workflow. All methods accept and/or return batches.
type RegistryPayload ยถ
type RegistryPayload[T Handle | UUID] struct { // RegistryTable is the table (or namespace) where the virtual IDs are stored or fetched. RegistryTable string // BlobTable is the paired blob table (or base filesystem path) used during Rollback and Commit. BlobTable string // CacheDuration specifies Redis cache duration. CacheDuration time.Duration // IsCacheTTL enables Redis TTL (sliding expiration) semantics when true. IsCacheTTL bool // IDs contains the virtual IDs (or Handles) to manage. IDs []T }
RegistryPayload represents a request/response payload to manage or fetch Handles/UUIDs in a registry table. T can be either Handle (for writes) or UUID (for reads/deletes).
func ExtractLogicalIDs ยถ
func ExtractLogicalIDs(storeHandles []RegistryPayload[Handle]) []RegistryPayload[UUID]
ExtractLogicalIDs converts a slice of RegistryPayload[Handle] to RegistryPayload[UUID] by mapping LogicalID.
type Relation ยถ
type Relation struct {
SourceFields []string `json:"source_fields"`
TargetStore string `json:"target_store"`
TargetFields []string `json:"target_fields"`
}
Relation describes a foreign key relationship to another store.
type ResourceAccess ยถ
type SchemaInferenceResult ยถ
type SchemaInferenceResult struct {
// Flat schema without prefixes for LLM correlation with Relations
Schema map[string]string
// Fields that belong to the Key
KeyFields []string
// Fields that belong to the Value
ValueFields []string
}
SchemaInferenceResult contains flat schema with field lists for LLM understanding.
func InferSchemaFromTypes ยถ
func InferSchemaFromTypes(key any, value any) SchemaInferenceResult
InferSchemaFromTypes uses reflection to directly inspect the types of key and value. Returns flat schema format without prefixes for better LLM understanding and correlation with Relations.
type SinglePhaseTransaction ยถ
type SinglePhaseTransaction struct {
SopPhaseCommitTransaction TwoPhaseCommitTransaction
// contains filtered or unexported fields
}
SinglePhaseTransaction wraps a TwoPhaseCommitTransaction providing an end-user friendly API and optional participation of other two-phase commit transactions.
func (*SinglePhaseTransaction) AddPhasedTransaction ยถ
func (t *SinglePhaseTransaction) AddPhasedTransaction(otherTransaction ...TwoPhaseCommitTransaction)
AddPhasedTransaction registers additional two-phase commit participants.
func (*SinglePhaseTransaction) Begin ยถ
func (t *SinglePhaseTransaction) Begin(ctx context.Context) error
Begin starts the wrapped transaction and any registered participants.
func (*SinglePhaseTransaction) Close ยถ
func (t *SinglePhaseTransaction) Close() error
Close calls Close on the wrapped transaction implementation.
func (*SinglePhaseTransaction) Commit ยถ
func (t *SinglePhaseTransaction) Commit(ctx context.Context) error
Commit executes phase 1 on all participants and then phase 2; on error, Rollback is invoked.
func (*SinglePhaseTransaction) CommitMaxDuration ยถ
func (t *SinglePhaseTransaction) CommitMaxDuration() time.Duration
CommitMaxDuration returns the configured commit duration cap from the underlying implementation.
func (*SinglePhaseTransaction) GetID ยถ
func (t *SinglePhaseTransaction) GetID() UUID
GetID returns the transaction ID.
func (*SinglePhaseTransaction) GetMode ยถ
func (t *SinglePhaseTransaction) GetMode() TransactionMode
GetMode returns the transaction mode.
func (*SinglePhaseTransaction) GetPhasedTransaction ยถ
func (t *SinglePhaseTransaction) GetPhasedTransaction() TwoPhaseCommitTransaction
GetPhasedTransaction returns the wrapped two-phase commit transaction.
func (*SinglePhaseTransaction) GetStores ยถ
func (t *SinglePhaseTransaction) GetStores(ctx context.Context) ([]string, error)
GetStores delegates to the wrapped transaction to list available stores.
func (*SinglePhaseTransaction) HasBegun ยถ
func (t *SinglePhaseTransaction) HasBegun() bool
HasBegun reports whether the transaction has started.
type StoreCacheConfig ยถ
type StoreCacheConfig struct {
// RegistryCacheDuration controls caching for registry objects.
RegistryCacheDuration time.Duration `json:"registry_cache_duration"`
// IsRegistryCacheTTL enables sliding TTL for registry cache.
IsRegistryCacheTTL bool `json:"is_registry_cache_ttl"`
// NodeCacheDuration controls caching for nodes.
NodeCacheDuration time.Duration `json:"node_cache_duration"`
// IsNodeCacheTTL enables sliding TTL for node cache.
IsNodeCacheTTL bool `json:"is_node_cache_ttl"`
// ValueDataCacheDuration controls caching for the item Value part when globally cached.
ValueDataCacheDuration time.Duration `json:"value_data_cache_duration"`
// IsValueDataCacheTTL enables sliding TTL for value data cache.
IsValueDataCacheTTL bool `json:"is_value_data_cache_ttl"`
// StoreInfoCacheDuration controls caching for StoreInfo records.
StoreInfoCacheDuration time.Duration `json:"store_info_cache_duration"`
// IsStoreInfoCacheTTL enables sliding TTL for store info cache.
IsStoreInfoCacheTTL bool `json:"is_store_info_cache_ttl"`
}
StoreCacheConfig declares cache durations and TTL flags for store artifacts.
func GetDefaultCacheConfig ยถ
func GetDefaultCacheConfig() StoreCacheConfig
GetDefaultCacheConfig returns the global default cache configuration.
func NewStoreCacheConfig ยถ
func NewStoreCacheConfig(cacheDuration time.Duration, isCacheTTL bool) *StoreCacheConfig
NewStoreCacheConfig returns a StoreCacheConfig with uniform cache durations and TTL settings applied. If cacheDuration is between 1ns and 5 minutes, it will be clamped to 5 minutes. TTL is disabled when duration is zero.
type StoreInfo ยถ
type StoreInfo struct {
// Name is the short store name.
Name string `json:"name" minLength:"1" maxLength:"128"`
// SlotLength is the number of items per node.
SlotLength int `json:"slot_length" min:"2" max:"20000"`
// IsUnique enforces uniqueness on the key of key/value items.
IsUnique bool `json:"is_unique"`
// Description optionally describes the store.
Description string `json:"description" maxLength:"1000"`
// RegistryTable is the registry table name.
RegistryTable string `json:"registry_table" minLength:"1" maxLength:"140"`
// BlobTable defines the target Erasure Coding (EC) configuration to use.
// The Database Options contain an EC configuration map keyed by a name, and
// this field's value is used to look up a match. If no matching entry is found,
// it falls back to the default EC config (which uses an empty string key "").
BlobTable string `json:"blob_table" minLength:"1" maxLength:"300"`
// RootNodeID is the root B-Tree node identifier.
RootNodeID UUID `json:"root_node_id"`
// Count is the total number of items persisted.
Count int64 `json:"count"`
// CountDelta is used internally to reconcile Count updates; it should not be persisted.
CountDelta int64 `json:"-"`
// Timestamp is the add/update time in milliseconds.
Timestamp int64 `json:"timestamp"`
// IsValueDataInNodeSegment stores the Value within the node segment when true.
IsValueDataInNodeSegment bool `json:"is_value_data_in_node_segment"`
// IsValueDataActivelyPersisted persists Value separately on Add/Update when true.
IsValueDataActivelyPersisted bool `json:"is_value_data_actively_persisted"`
// IsValueDataGloballyCached enables Redis caching of Value data when true.
IsValueDataGloballyCached bool `json:"is_value_data_globally_cached"`
// LeafLoadBalancing enables distribution to sibling nodes when capacity allows.
LeafLoadBalancing bool `json:"leaf_load_balancing"`
// CacheConfig overrides global cache settings per store.
CacheConfig StoreCacheConfig `json:"cache_config"`
// MapKeyIndexSpecification contains a CEL or index specification used by the comparer.
MapKeyIndexSpecification string `json:"mapkey_index_spec"`
// CELexpression specifies the CEL expression used as comparer for keys.
CELexpression string `json:"cel_expression,omitempty"`
// IsPrimitiveKey hints the Python binding which JSON B-Tree to instantiate on open.
// This is an internal feature and only needed to be managed by code when using (dynamic typed) languages like Python.
IsPrimitiveKey bool `json:"is_primitive_key"`
// Relations describes foreign key relationships to other stores.
Relations []Relation `json:"relations,omitempty"`
// Schema stores field types without prefixes for LLM correlation with Relations.
// Format: {"key": "string", "first_name": "string", "age": "number"}
Schema map[string]string `json:"schema,omitempty"`
// KeyFields lists the field names that are part of the Key.
// Example: ["key"] for primitive keys, ["id", "timestamp"] for composite keys
KeyFields []string `json:"key_fields,omitempty"`
// ValueFields lists the field names that are part of the Value.
// Example: ["first_name", "age", "email"] for the value object
ValueFields []string `json:"value_fields,omitempty"`
// CustomData stores optional arbitrary configuration for the store.
// It is intentionally flexible for integration-specific extensions.
CustomData map[string]any `json:"custom_data,omitempty"`
// For internal use only. Code can use this as hint.
NeedsMetaDataSave bool `json:"-"`
// Version allows versioning of the store info payload for future upgrades.
Version string `json:"version,omitempty"`
}
StoreInfo describes a B-Tree store configuration and runtime state persisted in the backend.
func NewStoreInfo ยถ
func NewStoreInfo(si StoreOptions) *StoreInfo
NewStoreInfo creates and normalizes a StoreInfo based on StoreOptions, applying default naming and cache policy.
func (*StoreInfo) DeleteCustomData ยถ
DeleteCustomData removes a value for the given key from CustomData.
func (*StoreInfo) GetCustomData ยถ
GetCustomData returns the value for the given key from CustomData.
func (*StoreInfo) GetCustomDataMap ยถ
GetCustomDataMap returns a copy of the CustomData map.
func (StoreInfo) IsCompatible ยถ
IsCompatible reports whether two StoreInfo configurations are compatible for merge/attach semantics.
func (StoreInfo) IsEmpty ยถ
IsEmpty reports whether the StoreInfo has zero values; an empty StoreInfo means the B-Tree does not yet exist.
func (*StoreInfo) SetCustomData ยถ
SetCustomData stores a value under the given key in CustomData.
func (*StoreInfo) SetCustomDataMap ยถ
SetCustomDataMap replaces the full CustomData map with a copy.
func (*StoreInfo) ValueDataSize ยถ
func (si *StoreInfo) ValueDataSize() ValueDataSize
Interpolates back into ValueDataSize
type StoreOptions ยถ
type StoreOptions struct {
// Name is the short name of the store.
Name string
// SlotLength is the number of items that can be stored in a node.
SlotLength int
// IsUnique enforces uniqueness on keys.
IsUnique bool
// IsValueDataInNodeSegment stores Value data within the B-Tree node segment when true.
// Smaller Value data benefits from this for locality; bigger data should be stored separately.
// If true, IsValueDataActivelyPersisted and IsValueDataGloballyCached are ignored.
IsValueDataInNodeSegment bool
// IsValueDataActivelyPersisted persists Value data to a separate partition on Add/Update and expects
// IsValueDataInNodeSegment to be false.
IsValueDataActivelyPersisted bool
// IsValueDataGloballyCached enables Redis caching for Value data when IsValueDataInNodeSegment is false.
IsValueDataGloballyCached bool
// LeafLoadBalancing allows distributing items to sibling nodes when there is capacity to avoid splits.
LeafLoadBalancing bool
// Description is an optional text describing the store.
Description string
// BlobStoreBaseFolderPath specifies a base folder path when using the filesystem blob store.
BlobStoreBaseFolderPath string
// DisableBlobStoreFormatting uses the store name directly as the blob store name (useful for S3-like systems).
DisableBlobStoreFormatting bool
// DisableRegistryStoreFormatting uses the store name directly as the registry store name.
DisableRegistryStoreFormatting bool
// CacheConfig overrides global cache durations and TTL behavior per store.
CacheConfig *StoreCacheConfig
// CELexpression specifies the CEL expression used as comparer for keys.
CELexpression string
// MapKeyIndexSpecification contains a CEL or index specification used by the comparer.
MapKeyIndexSpecification string
// IsPrimitiveKey hints Python bindings which JSON B-Tree type to instantiate during Open.
IsPrimitiveKey bool
// Relations describes foreign key-like relationships.
Relations []Relation
// CustomData stores optional arbitrary configuration for the store.
CustomData map[string]any
}
StoreOptions contains configuration fields used when creating a B-Tree store.
func ConfigureStore ยถ
func ConfigureStore(storeName string, uniqueKey bool, slotLength int, description string, valueDataSize ValueDataSize, blobStoreBaseFolderPath string) StoreOptions
ConfigureStore returns StoreOptions tuned according to the expected ValueDataSize. Choose carefully: mismatched size can hurt performance by over/under caching or persisting. blobStoreBaseFolderPath is used only for filesystem blob storage as a base directory.
type StoreRepository ยถ
type StoreRepository interface {
// Get retrieves store info by name(s).
Get(context.Context, ...string) ([]StoreInfo, error)
// GetWithTTL retrieves store info using TTL/sliding cache semantics.
GetWithTTL(context.Context, bool, time.Duration, ...string) ([]StoreInfo, error)
// GetAll lists all store names available in the backend.
GetAll(context.Context) ([]string, error)
// Add creates new store info entries and related tables (registry/blob).
Add(context.Context, ...StoreInfo) error
// Remove deletes store info by name and drops related tables.
Remove(context.Context, ...string) error
// Update modifies store info and reconciles Count using CountDelta.
Update(context.Context, []StoreInfo) ([]StoreInfo, error)
// Replicate performs post-commit replication of updated data managed by the repository.
Replicate(context.Context, []StoreInfo) error
}
StoreRepository specifies CRUD and replication methods for StoreInfo records.
type TaskRunner ยถ
type TaskRunner struct {
// contains filtered or unexported fields
}
TaskRunner is a thin wrapper around errgroup.Group that carries a context for convenience. Consider using errgroup directly in new code.
func NewTaskRunner ยถ
func NewTaskRunner(ctx context.Context, maxThreadCount int) *TaskRunner
NewTaskRunner creates a new TaskRunner. maxThreadCount > 0 limits the number of concurrent goroutines.
func (*TaskRunner) GetContext ยถ
func (tr *TaskRunner) GetContext() context.Context
GetContext returns the TaskRunner's context.
func (*TaskRunner) Go ยถ
func (tr *TaskRunner) Go(task func() error)
Go runs the provided task function in a new goroutine managed by the underlying errgroup.
func (*TaskRunner) Wait ยถ
func (tr *TaskRunner) Wait() error
Wait waits for all launched tasks to complete and returns the first encountered error, if any.
type Transaction ยถ
type Transaction interface {
// Begin starts the transaction.
Begin(ctx context.Context) error
// Commit finalizes the transaction.
Commit(ctx context.Context) error
// Rollback aborts the transaction.
Rollback(ctx context.Context) error
// HasBegun reports whether the transaction has started.
HasBegun() bool
// GetPhasedTransaction returns the underlying two-phase commit transaction for orchestration with other systems.
GetPhasedTransaction() TwoPhaseCommitTransaction
// AddPhasedTransaction registers external two-phase commit participants.
AddPhasedTransaction(otherTransaction ...TwoPhaseCommitTransaction)
// GetStores lists all available B-Tree stores from the backend.
GetStores(ctx context.Context) ([]string, error)
// Close releases any resources associated with the transaction.
Close() error
// GetID returns the transaction ID.
GetID() UUID
// CommitMaxDuration returns the configured maximum duration for commit operations.
// Effective runtime limit is min(ctx deadline, CommitMaxDuration()).
CommitMaxDuration() time.Duration
// OnCommit registers a callback to be executed after a successful commit.
OnCommit(callback func(ctx context.Context) error)
}
Transaction defines end-user-facing transactional operations.
func NewTransaction ยถ
func NewTransaction(mode TransactionMode, twoPhaseCommitTrans TwoPhaseCommitTransaction) (Transaction, error)
NewTransaction constructs a Transaction wrapper around a TwoPhaseCommitTransaction. mode controls permissions. When logging is true, lower layers may record commit steps to aid recovery and cleanup of expired resources.
type TransactionLog ยถ
type TransactionLog interface {
// PriorityLog returns the priority logger implementation.
PriorityLog() TransactionPriorityLog
// Add appends a transaction log entry.
Add(ctx context.Context, tid UUID, commitFunction int, payload []byte) error
// Remove deletes all logs for a transaction.
Remove(ctx context.Context, tid UUID) error
// GetOne returns the oldest hour bucket (older than 1 hour) and its logs for cleanup distribution.
GetOne(ctx context.Context) (UUID, string, []KeyValuePair[int, []byte], error)
// GetOneOfHour returns the available cleanup logs for a specific hour bucket.
GetOneOfHour(ctx context.Context, hour string) (UUID, []KeyValuePair[int, []byte], error)
// NewUUID generates a UUID suitable for the logging backend (e.g., time-based in Cassandra).
NewUUID() UUID
}
TransactionLog persists transaction steps and provides job-distribution accessors for cleanup tasks.
type TransactionMode ยถ
type TransactionMode int
TransactionMode enumerates the supported transaction behaviors.
const ( // NoCheck disallows any changes and skips read-version checks during commit. NoCheck TransactionMode = iota // ForWriting allows modifications to B-Tree stores within the transaction. ForWriting // ForReading disallows modifications; read-only. ForReading )
type TransactionOptions ยถ
type TransactionOptions struct {
// StoresFolders specifies the folders for replication.
StoresFolders []string `json:"stores_folders,omitempty"`
// ErasureConfig specifies the erasure coding configuration for replication.
ErasureConfig map[string]ErasureCodingConfig `json:"erasure_config,omitempty"`
// Keyspace to be used for the transaction (Cassandra).
Keyspace string `json:"keyspace,omitempty"`
// CacheType specifies the type of cache to use (e.g. InMemory, Redis).
CacheType L2CacheType `json:"cache_type"`
// RedisConfig specifies the Redis configuration when CacheType is Redis.
RedisConfig *RedisCacheConfig `json:"redis_config,omitempty"`
// Registry hash modulo value used for hashing.
RegistryHashModValue int `json:"registry_hash_mod,omitempty"`
// Transaction Mode can be Read-only or Read-Write.
Mode TransactionMode `json:"mode"`
// Transaction maximum "commit" time. Acts as the commit window cap and lock TTL.
MaxTime time.Duration `json:"max_time"`
}
func (TransactionOptions) GetDatabaseOptions ยถ
func (to TransactionOptions) GetDatabaseOptions() DatabaseOptions
GetDatabaseOptions returns the DatabaseOptions subset from TransactionOptions.
func (TransactionOptions) IsCassandraHybrid ยถ
func (to TransactionOptions) IsCassandraHybrid() bool
func (TransactionOptions) IsReplicated ยถ
func (to TransactionOptions) IsReplicated() bool
type TransactionPriorityLog ยถ
type TransactionPriorityLog interface {
// IsEnabled reports whether priority logging is enabled.
IsEnabled() bool
// Add appends a priority log for a transaction.
Add(ctx context.Context, tid UUID, payload []byte) error
// Remove deletes priority log file of a transaction.
Remove(ctx context.Context, tid UUID) error
// Get retrieves priority log details for a transaction.
Get(ctx context.Context, tid UUID) ([]RegistryPayload[Handle], error)
// GetBatch fetches up to batchSize of the oldest (older than 2 minutes) priority logs for processing.
GetBatch(ctx context.Context, batchSize int) ([]KeyValuePair[UUID, []RegistryPayload[Handle]], error)
// ProcessNewer iterates over all priority logs newer than 5 mins and invokes the processor callback for each.
ProcessNewer(ctx context.Context, processor func(tid UUID, payload []RegistryPayload[Handle]) error) error
// LogCommitChanges writes a special commit-change log used during drive reinstate for replication.
LogCommitChanges(ctx context.Context, stores []StoreInfo, newRootNodesHandles, addedNodesHandles, updatedNodesHandles, removedNodesHandles []RegistryPayload[Handle]) error
}
TransactionPriorityLog records prioritised transaction logs used for recovery and replication workflows.
type Tuple ยถ
type Tuple[T1 any, T2 any] struct { // First is the first element of the pair. First T1 // Second is the second element of the pair. Second T2 }
Tuple represents an ordered pair of two generic values when Key/Value semantics are not desired.
type TwoPhaseCommitTransaction ยถ
type TwoPhaseCommitTransaction interface {
// Begin starts the transaction.
Begin(ctx context.Context) error
// Phase1Commit performs the first phase (prepare) of the commit.
Phase1Commit(ctx context.Context) error
// Phase2Commit performs the second phase (finalize) of the commit.
Phase2Commit(ctx context.Context) error
// Rollback aborts the transaction and may be provided an error cause.
Rollback(ctx context.Context, err error) error
// HasBegun reports whether the transaction has started.
HasBegun() bool
// GetMode returns the configured TransactionMode.
GetMode() TransactionMode
// GetStores lists all available B-Tree stores from the backend.
GetStores(ctx context.Context) ([]string, error)
// Close releases any resources associated with the transaction implementation.
Close() error
// GetID returns the transaction ID.
GetID() UUID
// CommitMaxDuration returns the configured maximum duration for commit operations.
// Effective runtime limit is min(ctx deadline, CommitMaxDuration()).
CommitMaxDuration() time.Duration
// OnCommit registers a callback to be executed after a successful commit.
OnCommit(callback func(ctx context.Context) error)
}
TwoPhaseCommitTransaction defines infrastructure-facing two-phase commit operations.
type UICapability ยถ
type UICapability string
UICapability represents the UI-friendly permission key (e.g., "can_edit", "can_delete")
const ( UICapabilityRead UICapability = "can_read" UICapabilityEdit UICapability = "can_edit" UICapabilityDelete UICapability = "can_delete" UICapabilityAISelect UICapability = "can_ai_select" )
func ActionToUICapability ยถ
func ActionToUICapability(action Action) UICapability
ActionToUICapability maps an internal Action to the UI consumable Capability string
type UUID ยถ
UUID is a thin wrapper over github.com/google/uuid.UUID to keep SOP decoupled from the external package.
var NilUUID UUID
NilUUID is the zero-value UUID.
func NewUUID ยถ
func NewUUID() UUID
NewUUID returns a new randomly generated UUID. It retries on error with a 1ms backoff up to 10 times and panics only if all attempts fail (which should never happen under normal conditions).
func ParseUUID ยถ
ParseUUID converts a string to a UUID. It returns an error if the input is not a valid UUID.
func (UUID) Compare ยถ
Compare compares two UUIDs and returns -1 if x < y, 1 if x > y, and 0 if they are equal.
func (UUID) MarshalText ยถ
MarshalText implements the encoding.TextMarshaler interface.
func (*UUID) UnmarshalText ยถ
UnmarshalText implements the encoding.TextUnmarshaler interface.
type ValueDataSize ยถ
type ValueDataSize int
ValueDataSize categorizes the expected size of Value data to guide configuration helpers.
const ( // SmallData indicates small Value data that can be stored within the node segment. SmallData ValueDataSize = iota // MediumData indicates medium Value data that should be stored in a separate segment. MediumData // BigData indicates large Value data stored separately, actively persisted and typically not globally cached. BigData )
type Visibility ยถ
type Visibility string
const ( VisibilityPublic Visibility = "public" VisibilityPrivate Visibility = "private" VisibilitySystem Visibility = "system" )
Source Files
ยถ
Directories
ยถ
| Path | Synopsis |
|---|---|
|
adapters
|
|
|
cassandra
module
|
|
|
nats
module
|
|
|
redis
module
|
|
|
ai
module
|
|
|
bindings
|
|
|
main
command
|
|
|
Package btree provides the core B-Tree data structure and algorithms used by SOP.
|
Package btree provides the core B-Tree data structure and algorithms used by SOP. |
|
Package cache contains in-process MRU/L1 cache implementations and utilities used by SOP.
|
Package cache contains in-process MRU/L1 cache implementations and utilities used by SOP. |
|
Package cel provides a small wrapper around CEL expression compilation and evaluation used to compare map-based keys within SOP components.
|
Package cel provides a small wrapper around CEL expression compilation and evaluation used to compare map-based keys within SOP components. |
|
cmd
|
|
|
sop-a2a-agent
command
Command sop-a2a-agent serves tools/a2aagent over HTTP, registering the example db-maintenance runbook (tools/runbookstore.DBMaintenanceWorkflow) so an A2A client has something real to delegate a task against out of the box.
|
Command sop-a2a-agent serves tools/a2aagent over HTTP, registering the example db-maintenance runbook (tools/runbookstore.DBMaintenanceWorkflow) so an A2A client has something real to delegate a task against out of the box. |
|
sop-a2a-bridge
command
Command sop-a2a-bridge runs tools/a2abridge over stdio: an MCP server that resolves a running sop-a2a-agent's card and translates Claude's execute_step tool calls into real A2A task delegation against it.
|
Command sop-a2a-bridge runs tools/a2abridge over stdio: an MCP server that resolves a running sop-a2a-agent's card and translates Claude's execute_step tool calls into real A2A task delegation against it. |
|
sop-daemon
command
Command sop-daemon runs a loopback-only helper that executes shell commands on behalf of the SOP web UI (tools/httpserver serves a page that posts here after a human approves a proposed command).
|
Command sop-daemon runs a loopback-only helper that executes shell commands on behalf of the SOP web UI (tools/httpserver serves a page that posts here after a human approves a proposed command). |
|
sop-mcp-server
command
Command sop-mcp-server runs tools/mcpserver over stdio, serving the example db-maintenance runbook (tools/runbookstore.DBMaintenanceWorkflow) so an MCP client has something real to call read_sop/validate_step/ execute_step against out of the box.
|
Command sop-mcp-server runs tools/mcpserver over stdio, serving the example db-maintenance runbook (tools/runbookstore.DBMaintenanceWorkflow) so an MCP client has something real to call read_sop/validate_step/ execute_step against out of the box. |
|
testpb
command
|
|
|
Package common contains shared transaction and B-tree management helpers used by SOP.
|
Package common contains shared transaction and B-tree management helpers used by SOP. |
|
Package encoding provides pluggable marshal/unmarshal helpers used by SOP.
|
Package encoding provides pluggable marshal/unmarshal helpers used by SOP. |
|
Package main demonstrates Gemini 3.1 Pro optimizations integration.
|
Package main demonstrates Gemini 3.1 Pro optimizations integration. |
|
agent_memory
command
Package main demonstrates SOP as a durable memory engine for distributed AI agent swarms.
|
Package main demonstrates SOP as a durable memory engine for distributed AI agent swarms. |
|
interop_indexes
command
|
|
|
interop_jsondb
command
|
|
|
knowledgebase_cli
command
|
|
|
multi_redis
command
|
|
|
multi_redis_url
command
|
|
|
quickstart
command
Quickstart: the smallest possible SOP program.
|
Quickstart: the smallest possible SOP program. |
|
relations_demo
command
|
|
|
swarm_clustered
command
|
|
|
swarm_standalone
command
|
|
|
verify_barrier
command
Package main demonstrates ai/verify's barrier certificate blocking an out-of-order operational step in real time, the exact scenario described in docs/MCP_A2A_AND_VERIFICATION_ENGINE.md: an agent must not be allowed to drop the production database before a backup has been taken and validated, no matter what it claims about its own prior actions.
|
Package main demonstrates ai/verify's barrier certificate blocking an out-of-order operational step in real time, the exact scenario described in docs/MCP_A2A_AND_VERIFICATION_ENGINE.md: an agent must not be allowed to drop the production database before a backup has been taken and validated, no matter what it claims about its own prior actions. |
|
verify_barrier_nats
command
Package main is examples/verify_barrier plus one opt-in addition: adapters/nats.VerifyBridge, publishing every barrier decision to NATS so an external subscriber (this program's own second goroutine, standing in for a dashboard or another service) can observe them in real time.
|
Package main is examples/verify_barrier plus one opt-in addition: adapters/nats.VerifyBridge, publishing every barrier decision to NATS so an external subscriber (this program's own second goroutine, standing in for a dashboard or another service) can observe them in real time. |
|
Package fs contains filesystem-backed implementations used by SOP.
|
Package fs contains filesystem-backed implementations used by SOP. |
|
erasure
The decoder reverses the process done by "encoder.go"
|
The decoder reverses the process done by "encoder.go" |
|
incfs
module
|
|
|
infs
module
|
|
|
Package inmemory provides in-memory implementations of selected SOP backends, primarily for tests and lightweight scenarios.
|
Package inmemory provides in-memory implementations of selected SOP backends, primarily for tests and lightweight scenarios. |
|
internal
|
|
|
inredck
Package inredck contains the common kernel for Redis-based SOP implementations.
|
Package inredck contains the common kernel for Redis-based SOP implementations. |
|
logsafe
Package logsafe strips characters a caller could use to forge extra log lines out of values before they're written into a log message.
|
Package logsafe strips characters a caller could use to forge extra log lines out of values before they're written into a log message. |
|
netguard
Package netguard guards outbound HTTP requests and redirect targets against a caller-supplied URL steering this process somewhere it shouldn't go: the internal network, the cloud metadata endpoint, or an off-origin host for a redirect a user is expected to trust.
|
Package netguard guards outbound HTTP requests and redirect targets against a caller-supplied URL steering this process somewhere it shouldn't go: the internal network, the cloud metadata endpoint, or an off-origin host for a redirect a user is expected to trust. |
|
pathsafety
Package pathsafety guards destructive filesystem operations against a catastrophic path value.
|
Package pathsafety guards destructive filesystem operations against a catastrophic path value. |
|
jsondb
module
|
|
|
search
module
|
|
|
Package streamingdata is part of the internal baseline and is unsupported for public use.
|
Package streamingdata is part of the internal baseline and is unsupported for public use. |
|
tools
|
|
|
a2aagent
Package a2aagent exposes the same SOP runbook execution capability as tools/mcpserver, over the Agent2Agent (A2A) protocol instead of MCP: a task-delegation shape for orchestrators that speak A2A rather than MCP's tool-call shape.
|
Package a2aagent exposes the same SOP runbook execution capability as tools/mcpserver, over the Agent2Agent (A2A) protocol instead of MCP: a task-delegation shape for orchestrators that speak A2A rather than MCP's tool-call shape. |
|
a2abridge
Package a2abridge lets an MCP client (Claude Desktop, Claude Code, or any other MCP-speaking agent) drive a runbook served over Agent2Agent (A2A) by tools/a2aagent, without speaking A2A itself.
|
Package a2abridge lets an MCP client (Claude Desktop, Claude Code, or any other MCP-speaking agent) drive a runbook served over Agent2Agent (A2A) by tools/a2aagent, without speaking A2A itself. |
|
benchmark
command
|
|
|
healthcheck
command
Command healthcheck hits a local HTTP endpoint and exits 0 on a 2xx response, non-zero otherwise.
|
Command healthcheck hits a local HTTP endpoint and exits 0 on a 2xx response, non-zero otherwise. |
|
httpserver
command
|
|
|
mcpserver
Package mcpserver exposes SOP runbooks to MCP clients (an LLM agent, an orchestration framework, another service) as three tools: read_sop, validate_step, and execute_step.
|
Package mcpserver exposes SOP runbooks to MCP clients (an LLM agent, an orchestration framework, another service) as three tools: read_sop, validate_step, and execute_step. |
|
runbookstore
Package runbookstore holds the registered ai/verify.Workflow runbooks and their in-progress execution traces, shared by every protocol front-end this repo exposes them through (tools/mcpserver, tools/a2aagent).
|
Package runbookstore holds the registered ai/verify.Workflow runbooks and their in-progress execution traces, shared by every protocol front-end this repo exposes them through (tools/mcpserver, tools/a2aagent). |