ledger

package module
v0.0.20 Latest Latest
Warning

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

Go to latest
Published: Jul 17, 2026 License: MIT Imports: 19 Imported by: 6

README

Ledger — Distributed State Directory

go get github.com/bbmumford/ledger

Related: Aether (wire protocol) · Whisper (gossip engine)

Ledger is an append-only, signed-record distributed state directory for peer meshes. It provides typed record storage, materialised caches, pluggable persistence backends (memory / SQLite / Turso), signed anchor snapshots for fast bootstrap, and a distributed blob storage pipeline with quota / pinning / replication / rarest-first fetch. Designed to ride on top of any wire transport — pairs naturally with Aether (Noise-UDP mesh protocol) and Whisper (directory gossip).


Status

Area Status
Record schema + types Stable
DirectoryCache (materialised views) Stable
Pluggable CacheStore (memory / SQLite / Turso) Stable
Append-only ledger + subscriber broadcast Stable
Anchor snapshot + signing Stable
Distributed blob pipeline (chunker, fetcher, replicator, rarest-first, endgame, pinning, quota, abuse, proof, resolver) Stable
Whisper adapter (Record → GossipPayload) Stable
v0.0.1 release Shipped (2026-04-20)

In production use by the HSTLES mesh across bootstrap.hstles.com, node.hstles.com, and every ORBTR endpoint ({devices,relay,app,get,help,node,orbtr}.orbtr.io) — full fleet currently carries ~500 directory records per region with anchor snapshots rotating every 5 minutes.


What Ledger Is

Ledger sits between a wire protocol (Aether) and applications:

Application (HSTLES mesh, ORBTR agent, custom app)
  ↓ queries records, applies mutations, subscribes to topic updates
Ledger (state directory)
  ↓ serialises records as opaque gossip payload
Aether + Whisper (transport + epidemic propagation)

A transport moves bytes. Ledger gives those bytes meaning — typed records for peers, roles, reachability, latency, capabilities, and blob references. Whisper's gossip engine carries Ledger records between peers using delta sync, digest comparison, and per-peer watermarks.


Components

Component Path Description
Record types types.go Record, Topic, MemberRecord, RoleRecord, ReachRecord, LatencyRecord, HandlerMetadata, KeyOpsRecord, QuorumRecord
Cache cache/ DirectoryCache — materialised view with per-topic queries (Members, Roles, Reach, Latency); CRDT semantics
CacheStore cache/store.go, cache/store_memory.go Pluggable backend interface (memory built-in; SQLite/Turso via consumer)
Ledger storage/ledger.go, storage/local.go, storage/memory.go Append-only record log with subscriber broadcast
Anchor anchor/ Signed snapshot of all records for fast bootstrap (generator.go, crypto.go, types.go)
Blob blob/ + blob/distributed/ Content-addressed blob store + distributed pipeline: chunker, encryption, fetcher, advertiser, replicator, rarest-first, endgame, proof (on-chain + local), pinning, quota, abuse guard
Whisper adapter whisper_adapter.go Implements Whisper's GossipPayload interface for Ledger records
Metrics metrics.go Prometheus-compatible counters + gauges
ACL acl.go Topic-level write authorisation
Compaction compaction.go Tombstone + expiry compaction on the append log
Conflict conflict.go Lamport-clock + sequence conflict resolution

Record Schema

// Record is a single entry in the distributed directory.
type Record struct {
    Topic        Topic           // member, role, reach, latency, keyops, quorum
    NodeID       string          // which peer this record describes
    TenantID     string          // tenant scope (empty = global)
    Body         json.RawMessage // typed payload (MemberRecord, RoleRecord, etc.)
    Timestamp    time.Time
    LamportClock uint64          // causal ordering
    Seq          uint32          // per-node sequence number
    Tombstone    bool            // soft delete
    ExpiresAt    time.Time       // TTL
    Signature    []byte          // Ed25519 signature over canonical encoding
}
Topics
Topic Record Type Description
member MemberRecord Peer identity, region, service name, public key, attributes
role RoleRecord Namespace-qualified roles (e.g., "platform.identity"), handler metadata
reach ReachRecord Advertised network addresses (IP, port, protocol, scope)
latency LatencyRecord Measured RTT between peers
keyops KeyOpsRecord Key rotation operations
quorum QuorumRecord Anchor quorum membership
blob — Content-addressed blob references (SHA-256 hash → source peers)

Architecture Principles

  1. Append-only. Every state change appends a new record. Conflicts resolved by Lamport clock + sequence; older records tombstoned, never mutated in place.
  2. Signed end-to-end. Every record carries an Ed25519 signature by its NodeID. A peer can verify any record came from the claimed author.
  3. CRDT-friendly. Records converge deterministically given the same set of observations — ordering doesn't matter for eventual consistency, Lamport clock breaks ties for causally-dependent updates.
  4. Transport-agnostic. Ledger doesn't import any network code. The whisper_adapter.go provides GossipPayload serialisation; consumers wire Ledger to Whisper (or any other gossip engine) at the integration layer.
  5. Pluggable persistence. CacheStore is an interface with a memory-backed default; consumers plug in SQLite, Turso, or any KV store that satisfies the contract.
  6. Tenant isolation. Every record carries a TenantID. Queries scope to a tenant; the empty string is the global scope. Multi-tenant deployments never see cross-tenant state.

API Docs

Generated API reference (pkgsite) for each tagged release is published to GitHub Pages:


Quick Start

import (
    "context"
    "time"

    "github.com/bbmumford/ledger"
    "github.com/bbmumford/ledger/cache"
    "github.com/bbmumford/ledger/storage"
)

// In-memory ledger + cache — no external dependencies.
ctx := context.Background()
log := storage.NewMemoryLedger()
store := cache.NewMemoryStore()
dc := cache.NewDirectoryCache(store)
log.Subscribe(dc)

// Append a member record.
rec := &ledger.Record{
    Topic:     ledger.TopicMember,
    NodeID:    "vl1_abc123",
    Timestamp: time.Now(),
    Body:      /* encoded MemberRecord */,
}
log.Append(ctx, rec)

// Query the materialised view.
members, _ := dc.Members(ctx, "", cache.MemberQuery{Region: "syd"})

License

MIT

Copyright (c) 2026 HSTLES / ORBTR Pty Ltd

Documentation

Overview

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

Package lad implements the Ledger-as-Directory abstraction for peer discovery and membership tracking within the Orbtr mesh.

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

  • Copyright (c) 2026 HSTLES / ORBTR Pty Ltd. All Rights Reserved.

  • Queries: licensing@hstles.com

Index

Constants

View Source
const ContentHashLen = 16

ContentHashLen is the size of a content hash in bytes (16 — half of SHA-256, sufficient at fleet scales up to ~10⁸ records).

View Source
const DefaultExpiryInterval = 60 * time.Second

DefaultExpiryInterval is the default period between expiry sweeps.

View Source
const ReachDeltaSchemaFlag uint16 = 0x8000

ReachDeltaSchemaFlag marks a ReachRecord body as a reach-layer delta (address ops) rather than a full snapshot. Matches reach.DeltaSchemaFlag.

Variables

View Source
var (
	ErrSignatureRequired = errors.New("ledger: record requires signature")
	ErrSignatureInvalid  = errors.New("ledger: invalid record signature")
)
View Source
var (
	// ErrACLDenied is returned when an ACL check rejects a record.
	ErrACLDenied = errors.New("ledger: record rejected by ACL")
)
View Source
var ErrBlobNotFound = errors.New("ledger: blob not found")

Functions

func BodyKey

func BodyKey(r Record) string

BodyKey derives the storage key from the record's Body (collection:key pattern).

func ComputeCID

func ComputeCID(data []byte) string

ComputeCID returns the SHA-256 content hash of data.

func IdentityViewKey added in v0.0.8

func IdentityViewKey(nodeID, topic string) string

IdentityViewKey returns the canonical key for an IdentityView, suitable for indexing in the cache projection map.

func NodeIDKey

func NodeIDKey(r Record) string

NodeIDKey derives the storage key from the record's NodeID.

func RecordFingerprint

func RecordFingerprint(rec Record) uint64

RecordFingerprint computes an FNV-1a fingerprint of a record for dedup.

func SignRecord

func SignRecord(rec *Record, privateKey ed25519.PrivateKey)

SignRecord signs a record's content with the given Ed25519 private key. Sets AuthorPubKey and Signature fields on the record. AuthorPubKey MUST be assigned before computing signatureContent — signatureContent reads rec.AuthorPubKey, so signing the record with an empty pubkey here produces bytes the receiver cannot reproduce (the wire carries the 32-byte pubkey, but the signed bytes would carry a 4-byte zero length). The ordering bug here is what every signed-topic record on the wire was failing verify against — VerifyRecord's reconstructed content always disagreed with the signer's content.

func VerifyRecord

func VerifyRecord(rec Record) bool

VerifyRecord checks the Ed25519 signature on a record.

Types

type ACLFunc

type ACLFunc func(topic string, authorPubKey []byte, record Record) error

ACLFunc checks whether a record is allowed to be applied to a topic. Called in Apply() before merge. Return nil to allow, error to reject.

type BlobStore

type BlobStore interface {
	Put(data []byte) (cid string, err error)
	Get(cid string) ([]byte, error)
	Has(cid string) (bool, error)
	Delete(cid string) error
	Size() (count int, bytes int64, err error)
}

BlobStore manages content-addressed binary objects alongside records. This is the historical (pre-Session-B) interface: no context, returns (count, bytes, err) from Size. Retained verbatim so existing callers (notably minerva/persistence/tiered.go and the HSTLES library) compile unchanged. New code should prefer ContextBlobStore.

type BlobStoreAdapter

type BlobStoreAdapter struct {
	Inner BlobStore
}

BlobStoreAdapter wraps a legacy BlobStore as a ContextBlobStore. The adapter ignores the context (the underlying store has no cancellation hook); callers that need cancellation should implement ContextBlobStore directly. Size returns only the byte total — the record count from the legacy API is discarded.

func NewBlobStoreAdapter

func NewBlobStoreAdapter(inner BlobStore) *BlobStoreAdapter

NewBlobStoreAdapter returns a ContextBlobStore backed by inner.

func (*BlobStoreAdapter) Delete

func (a *BlobStoreAdapter) Delete(_ context.Context, cid string) error

Delete delegates to the wrapped BlobStore (ctx is unused).

func (*BlobStoreAdapter) Get

func (a *BlobStoreAdapter) Get(_ context.Context, cid string) ([]byte, error)

Get delegates to the wrapped BlobStore (ctx is unused).

func (*BlobStoreAdapter) Has

func (a *BlobStoreAdapter) Has(_ context.Context, cid string) (bool, error)

Has delegates to the wrapped BlobStore (ctx is unused).

func (*BlobStoreAdapter) Put

func (a *BlobStoreAdapter) Put(_ context.Context, data []byte) (string, error)

Put delegates to the wrapped BlobStore (ctx is unused).

func (*BlobStoreAdapter) Size

func (a *BlobStoreAdapter) Size(_ context.Context) (int64, error)

Size returns the byte total from the legacy BlobStore, discarding the record count so the return shape matches ContextBlobStore.

type CacheStateStore

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

CacheStateStore wraps a DirectoryCache-like store to implement whisper.StateStore so the native G1 exchange handler can drive delta-sync and full-sync transfers without knowing about Record structure. Records travel as opaque bytes; the supplied marshal / unmarshal funcs bridge Record↔wire format at the adapter boundary.

func NewCacheStateStore

func NewCacheStateStore(
	cache interface {
		Fingerprint() uint64
		Dump() []Record
		DumpSince(since time.Time) []Record
		Apply(rec Record) error
	},
	marshal func(Record) ([]byte, error),
	unmarshal func([]byte) (Record, error),
) *CacheStateStore

NewCacheStateStore returns a StateStore adapter suitable for whisper.WithG1Store. marshal and unmarshal are called once per record transferred; callers typically use the same codec that their G1Codec uses on the envelope boundary.

func (*CacheStateStore) Apply

func (s *CacheStateStore) Apply(data []byte) error

Apply ingests a single inbound record from gossip or rumor push.

func (*CacheStateStore) Delta added in v0.0.7

func (s *CacheStateStore) Delta(since time.Time) [][]byte

Delta returns records mutated since the given time. A zero time falls back to Snapshot so whisper's handler never has to special- case the bootstrap path.

func (*CacheStateStore) Fingerprint

func (s *CacheStateStore) Fingerprint() uint64

Fingerprint returns the cache's order-independent fingerprint for G2 digest probes.

func (*CacheStateStore) Snapshot added in v0.0.7

func (s *CacheStateStore) Snapshot() [][]byte

Snapshot returns every live record as opaque bytes. Records that fail to marshal are dropped so a single malformed entry doesn't starve the rest of the payload.

type CausalWatermark

type CausalWatermark map[Topic]uint64

CausalWatermark tracks the last observed sequence per topic.

func (CausalWatermark) Copy

func (cw CausalWatermark) Copy() CausalWatermark

Copy returns a deep copy of the watermark map.

type CompactionPolicy

type CompactionPolicy struct {
	MaxTombstoneAge time.Duration // prune tombstones older than this (0 = keep all)
	MaxVersions     int           // keep last N versions per key (0 = unlimited)
	MaxRecords      int           // hard cap per topic (0 = unlimited)
}

CompactionPolicy controls how old records are pruned.

func DefaultCompactionPolicy

func DefaultCompactionPolicy() CompactionPolicy

DefaultCompactionPolicy returns a sensible default.

type ConflictLog

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

ConflictLog is a ring buffer of recent merge conflicts per topic.

func NewConflictLog

func NewConflictLog(maxEntries int) *ConflictLog

NewConflictLog creates a conflict log with the given capacity.

func (*ConflictLog) Clear

func (cl *ConflictLog) Clear()

Clear removes all logged conflicts.

func (*ConflictLog) Entries

func (cl *ConflictLog) Entries(topic Topic, limit int) []ConflictRecord

Entries returns all logged conflicts, optionally filtered by topic.

func (*ConflictLog) Record

func (cl *ConflictLog) Record(c ConflictRecord)

Record adds a conflict to the log.

func (*ConflictLog) SetEnabled

func (cl *ConflictLog) SetEnabled(enabled bool)

SetEnabled enables or disables conflict logging.

type ConflictRecord

type ConflictRecord struct {
	Topic     Topic
	Key       string
	Winner    Record
	Loser     Record
	Timestamp time.Time
	MergeFunc string // name of merge function that resolved it
}

ConflictRecord captures the details of a merge conflict resolution.

type ContentHash added in v0.0.8

type ContentHash [ContentHashLen]byte

ContentHash is the canonical content-addressed identifier for a record. Used by:

  • whisper.iblt for set reconciliation cell keys
  • whisper.RumorPusher for ACK identity (rumor ID = content hash)
  • whisper.ReconcileStore.FetchByID lookups
  • cache projection layers (IdentityView)

Deterministic across nodes given the same record content. Two records with identical (Topic, NodeID, Body, HLCTimestamp, Tombstone) produce the same hash. Signature is deliberately excluded — multiple valid signatures over the same content (e.g. after a key rotation) all produce the same hash, so dedup remains correct across rotations.

func SortedHashes added in v0.0.8

func SortedHashes(views []IdentityView) []ContentHash

SortedHashes returns the content hashes from a slice of IdentityViews in deterministic order. Used by reconciliation drivers that need a consistent IBLT input ordering across nodes.

type ContextBlobStore

type ContextBlobStore interface {
	Put(ctx context.Context, data []byte) (cid string, err error)
	Get(ctx context.Context, cid string) ([]byte, error)
	Has(ctx context.Context, cid string) (bool, error)
	Delete(ctx context.Context, cid string) error
	Size(ctx context.Context) (int64, error)
}

ContextBlobStore is the Session-B-harmonised blob-store interface shared with Minerva's messaging subsystem. It accepts a context on every method so long-running operations (IndexedDB writes, S3 PUTs, Turso range scans) can be cancelled; Size returns a single int64 byte total because callers that also want the record count can wrap Has with a counter. Adapt legacy BlobStore instances via BlobStoreAdapter.

type DebugLogger

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

DebugLogger provides conditional debug logging for Ledger subsystems.

func NewDebug

func NewDebug(name string) *DebugLogger

NewDebug creates a debug logger for the named subsystem.

func (*DebugLogger) Printf

func (d *DebugLogger) Printf(format string, args ...interface{})

Printf logs a formatted message if this subsystem is enabled.

type ExpirySource

type ExpirySource interface {
	PruneExpired() (removed int, err error)
}

ExpirySource is implemented by any cache that can prune TTL-expired records.

type ExpiryWorker

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

ExpiryWorker runs a background goroutine that prunes expired records.

func NewExpiryWorker

func NewExpiryWorker(cache ExpirySource, interval time.Duration) *ExpiryWorker

NewExpiryWorker creates an ExpiryWorker that calls cache.PruneExpired() on the given interval. If interval is zero, DefaultExpiryInterval is used.

func (*ExpiryWorker) PruneExpired

func (w *ExpiryWorker) PruneExpired() (int, error)

PruneExpired delegates to the underlying cache's PruneExpired method. Useful for on-demand pruning outside the background ticker loop.

func (*ExpiryWorker) Run

func (w *ExpiryWorker) Run(ctx context.Context)

Run starts the expiry loop and blocks until ctx is cancelled.

type HandlerMetadata

type HandlerMetadata struct {
	Name           string   `json:"name"`
	RequiresAuth   bool     `json:"requiresAuth"`
	RequiredScopes []string `json:"requiredScopes,omitempty"`
	TimeoutMs      int      `json:"timeoutMs"`
}

HandlerMetadata describes security requirements for a handler

type IdentityView added in v0.0.8

type IdentityView struct {
	NodeID      string
	Topic       string
	ContentHash ContentHash
	HLC         uint64
	UpdatedAt   int64 // unix nano; consumers convert to time.Time as needed
}

IdentityView is a cache-projection row representing a single (NodeID, Topic) pair's most recent record. Used by reconciliation cell generation so the IBLT operates over per-(node, topic) keys without requiring Member and Reach to unify on the wire.

Cache-only — never serialised to the wire. The wire format keeps Member and Reach as separate signed records per §5.27 of the 2026-04-23 reach publisher redesign.

type IndexFunc

type IndexFunc func(Record) []string

IndexFunc derives secondary index keys from a record. Returns zero or more index keys that will be maintained automatically.

type KeyFunc

type KeyFunc func(Record) string

KeyFunc derives the storage key from a record. Default is NodeIDKey (keyed by NodeID). CRDTs use BodyKey.

type LatencyRecord

type LatencyRecord struct {
	FromNode     string    `json:"fromNode"`
	ToNode       string    `json:"toNode"`
	RTTMs        int64     `json:"rttMs"`                  // Application RTT (gossip/data exchange latency)
	InitialRTTMs int64     `json:"initialRttMs,omitempty"` // Network RTT (connection setup latency)
	Transport    string    `json:"transport,omitempty"`    // "noise-udp" or "gossip-tls"
	MeasuredAt   time.Time `json:"measuredAt"`
	ExpiresAt    time.Time `json:"expiresAt"`
}

LatencyRecord stores a directional RTT measurement between two nodes. Published by the measuring node: "I measured Xms RTT to peer Y."

func UnmarshalLatency

func UnmarshalLatency(body []byte) (LatencyRecord, error)

UnmarshalLatency decodes a LatencyRecord from JSON.

type Ledger

type Ledger interface {
	Head(ctx context.Context) (CausalWatermark, error)
	Append(ctx context.Context, rec Record) error
	BatchAppend(ctx context.Context, records []Record) error
	Stream(ctx context.Context, from CausalWatermark, topics []Topic) (<-chan Record, error)
	Snapshot(ctx context.Context) (io.ReadCloser, error)
}

Ledger exposes append and replication operations for the directory log.

type MemberRecord

type MemberRecord struct {
	TenantID  string            `json:"tenant"`
	NodeID    string            `json:"nodeId"`
	PubKey    []byte            `json:"pubkey"`
	CreatedAt time.Time         `json:"createdAt"`
	ExpiresAt time.Time         `json:"expiresAt,omitempty"`
	Attrs     map[string]string `json:"attrs,omitempty"`
	Handlers  []HandlerMetadata `json:"handlers,omitempty"` // Handler security metadata
}

MemberRecord mirrors the MemberAdd log schema once decoded.

func UnmarshalMember

func UnmarshalMember(body []byte) (MemberRecord, error)

UnmarshalMember decodes a member record body.

type MemoryBlobStore

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

MemoryBlobStore is an in-memory BlobStore for testing.

func NewMemoryBlobStore

func NewMemoryBlobStore() *MemoryBlobStore

NewMemoryBlobStore creates a new in-memory blob store.

func (*MemoryBlobStore) Delete

func (s *MemoryBlobStore) Delete(cid string) error

func (*MemoryBlobStore) Get

func (s *MemoryBlobStore) Get(cid string) ([]byte, error)

func (*MemoryBlobStore) Has

func (s *MemoryBlobStore) Has(cid string) (bool, error)

func (*MemoryBlobStore) Put

func (s *MemoryBlobStore) Put(data []byte) (string, error)

func (*MemoryBlobStore) Size

func (s *MemoryBlobStore) Size() (int, int64, error)

type MergeFunc

type MergeFunc func(existing, incoming Record) (Record, error)

MergeFunc resolves conflicts between an existing record and an incoming record. Returns the winning record. Called by DirectoryCache.Apply().

type MetricsRecorder

type MetricsRecorder interface {
	RecordApplied(topic string, merged bool)
	RecordExpired(topic string)
	MergeConflict(topic string, winner string)
	TopicSize(topic string, count int, bytes int64)
	GossipExchange(topic string, sent, received int)
}

MetricsRecorder receives operational metrics from the cache.

type NoopMetricsRecorder

type NoopMetricsRecorder struct{}

NoopMetricsRecorder is a no-op implementation of MetricsRecorder. Use it as a default when no real metrics backend is configured.

func (NoopMetricsRecorder) GossipExchange

func (NoopMetricsRecorder) GossipExchange(topic string, sent, received int)

func (NoopMetricsRecorder) MergeConflict

func (NoopMetricsRecorder) MergeConflict(topic string, winner string)

func (NoopMetricsRecorder) RecordApplied

func (NoopMetricsRecorder) RecordApplied(topic string, merged bool)

func (NoopMetricsRecorder) RecordExpired

func (NoopMetricsRecorder) RecordExpired(topic string)

func (NoopMetricsRecorder) TopicSize

func (NoopMetricsRecorder) TopicSize(topic string, count int, bytes int64)

type QueryFilter

type QueryFilter struct {
	NodeID    string    // filter by node (empty = all)
	KeyPrefix string    // filter by key prefix (empty = all)
	Since     time.Time // records after this time (zero = all)
	Limit     int       // max results (0 = unlimited)
}

QueryFilter provides generic query parameters for any topic.

type ReachAddress

type ReachAddress struct {
	Host  string `json:"ip"`
	Port  int    `json:"port"`
	Proto string `json:"proto"`
	Scope string `json:"scope"` // public|lan|rfc1918
}

ReachAddress represents a single advertised endpoint.

func (ReachAddress) DialAddress

func (a ReachAddress) DialAddress() string

DialAddress returns the protocol-appropriate address for dialing. For WSS, this builds the full WebSocket URL. For other protocols, it returns host:port (same as String()).

func (ReachAddress) String

func (a ReachAddress) String() string

String returns host:port for dialing.

type ReachQuery

type ReachQuery struct {
	NodeID   string  // Filter by specific node ID (optional)
	Role     string  // Filter by role (optional)
	MinScore float64 // Minimum availability score (optional, 0.0-1.0)
	Limit    int     // Maximum number of results (optional)
}

ReachQuery provides filters for querying reachability records

type ReachRecord

type ReachRecord struct {
	TenantID      string         `json:"tenant"`
	NodeID        string         `json:"nodeId"`
	Seq           uint64         `json:"seq"`
	Addresses     []ReachAddress `json:"addrs"`
	Region        string         `json:"region"` // Hosting region (e.g., "iad", "us-east-1")
	NATType       string         `json:"natType"`
	NATObserved   string         `json:"natObserved"`
	LatencyMillis int64          `json:"latencyMs"`
	Availability  float64        `json:"availability"`
	LoadFactor    float64        `json:"loadFactor"` // 0.0 (idle) to 1.0 (overloaded)
	ExpiresAt     time.Time      `json:"expiresAt"`
	UpdatedAt     time.Time      `json:"ts"`

	// SchemaVersion lets the cache detect delta-encoded bodies (bit 15
	// set) and skip applying them — deltas are an address-change
	// optimisation at the reach layer and would otherwise clobber the
	// last signed full snapshot (wiping Metadata/Addresses/Region).
	// Full snapshots are re-emitted on a cadence, so Metadata stays
	// fresh regardless.
	SchemaVersion uint16 `json:"v,omitempty"`

	// Metadata is consumer-defined identity payload carried alongside
	// reachability. Examples: {"service_name":"devices.orbtr.io",
	// "roles":"anchor","region":"iad","fly_machine":"abc"} for mesh
	// nodes, or {"hostname":"alice-laptop","device_id":"dev-abc"} for
	// agents. The ledger cache derives MemberRecord views from this
	// map on read (see DirectoryCache.Members), so consumers don't need
	// a parallel Member publish path — a single signed ReachRecord is
	// the source of truth for both "how to reach me" and "who I am".
	Metadata map[string]string `json:"meta,omitempty"`
}

ReachRecord mirrors the ReachSet schema plus derived metrics.

func UnmarshalReach

func UnmarshalReach(body []byte) (ReachRecord, error)

UnmarshalReach decodes a record body into a ReachRecord.

func (ReachRecord) IsReachDelta added in v0.0.4

func (r ReachRecord) IsReachDelta() bool

IsReachDelta reports whether the body behind a ReachRecord is a delta.

func (ReachRecord) LatencyDuration

func (r ReachRecord) LatencyDuration() time.Duration

LatencyDuration exposes the latency as a time.Duration for callers.

type Record

type Record struct {
	Topic           Topic           `json:"topic"`
	Seq             uint64          `json:"seq"`
	TenantID        string          `json:"tenantId"`
	NodeID          string          `json:"nodeId"`
	Body            json.RawMessage `json:"body"`
	Signature       []byte          `json:"sig"`
	Timestamp       time.Time       `json:"ts"`
	LamportClock    uint64          `json:"lc,omitempty"`        // causal ordering
	HLCTimestamp    uint64          `json:"hlc,omitempty"`       // hybrid logical clock (0 = unused)
	Tombstone       bool            `json:"tombstone,omitempty"` // true = deletion marker
	DeletedAt       time.Time       `json:"deletedAt,omitempty"`
	TombstoneReason string          `json:"tombstoneReason,omitempty"` // "explicit", "cap"
	ExpiresAt       time.Time       `json:"expiresAt,omitempty"`       // TTL expiration (zero = never)
	AuthorPubKey    []byte          `json:"authorPubkey,omitempty"`    // Ed25519 public key of record author
	BlobCID         string          `json:"blobCid,omitempty"`         // content-addressed blob reference
}

Record is the canonical ledger entry.

func OverwriteMerge

func OverwriteMerge(existing, incoming Record) (Record, error)

OverwriteMerge is the default merge function. Latest timestamp or highest LamportClock wins.

func (Record) ContentHash added in v0.0.8

func (r Record) ContentHash() ContentHash

ComputeContentHash returns the canonical content hash for a record. Stable across encoder versions: hashes the canonicalised byte representation of (Topic, TenantID, NodeID, Body, HLC, Tombstone) — never the signature, never the receive timestamp.

The canonical form is fixed at:

[len(topic)|topic][len(tenant)|tenant][len(node)|node]
[len(body)|body][hlc_uint64][tombstone_byte]

All length prefixes are big-endian uint32. SHA-256 truncated to the first 16 bytes.

type RoleRecord

type RoleRecord struct {
	TenantID string            `json:"tenant"`
	NodeID   string            `json:"nodeId"`
	Roles    []string          `json:"roles"`
	Handlers []HandlerMetadata `json:"handlers,omitempty"` // Handler security metadata
	MaxGrade int               `json:"maxGrade,omitempty"` // Highest grade this node can support (0-4)
	Updated  time.Time         `json:"ts"`
}

RoleRecord mirrors the RoleSet schema.

func UnmarshalRole

func UnmarshalRole(body []byte) (RoleRecord, error)

UnmarshalRole decodes a role record body.

type Topic

type Topic string

Topic identifies a logical log shard within the ledger.

const (
	TopicMember  Topic = "member"
	TopicRole    Topic = "role"
	TopicReach   Topic = "reach"
	TopicKeyOps  Topic = "keyops"
	TopicQuorum  Topic = "quorum"
	TopicLatency Topic = "latency"
)

type TopicConfig

type TopicConfig struct {
	Merge            MergeFunc                                    // conflict resolution (nil = OverwriteMerge)
	Key              KeyFunc                                      // key derivation (nil = NodeIDKey)
	ExpiryEnabled    bool                                         // enable TTL expiration
	RequireSignature bool                                         // reject unsigned records
	VerifyFunc       func(pubkey, signature, content []byte) bool // custom signature verifier (nil = Ed25519)
	SignFunc         func(content []byte) (pubkey, sig []byte)    // for local writes
}

TopicConfig configures per-topic behavior.

Directories

Path Synopsis
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
blob
distributed
Package distributed is a transitional shell.
Package distributed is a transitional shell.
distributed/onchain
Package onchain holds the on-chain proof-of-storage + paid-pinning surface that the rest of the legacy ledger/blob/distributed package no longer needs.
Package onchain holds the on-chain proof-of-storage + paid-pinning surface that the rest of the legacy ledger/blob/distributed package no longer needs.
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
internal
bloom
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.
* Copyright (c) 2026 HSTLES / ORBTR Pty Ltd.

Jump to

Keyboard shortcuts

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