Documentation
¶
Overview ¶
Package object binds one any-sync object tree to one crdt.Controller and exposes the lifecycle surface used by space/: Create / Derive / Modify / Delete / Subscribe, plus ocache wiring so trees are loaded on demand and TTL-closed when idle.
Not caller-facing. Middleware never holds an Object handle — it operates at the space level and references objects by objectId. See docs/object.md.
Package object owns the change-payload wire codec, the per-space VersionId allocator, and the binder that ties one any-sync ObjectTree to one crdt.Controller. See doc.go for the broader role.
Index ¶
- Variables
- type AfterApply
- type AfterReplay
- type ApplyGate
- type Codec
- type Config
- type Object
- func (o *Object) ApplyDecoded(ctx context.Context, ch crdt.Change) error
- func (o *Object) Close() error
- func (o *Object) ColdRestore(ctx context.Context) error
- func (o *Object) Controller() *crdt.Controller
- func (o *Object) Id() string
- func (o *Object) InjectedSet(ctx context.Context, ch crdt.Change) (WriteResult, error)
- func (o *Object) LocalSet(ctx context.Context, ch crdt.Change) (WriteResult, error)
- func (o *Object) LocalWrite(ctx context.Context, ch crdt.Change) (WriteResult, error)
- func (o *Object) Rebuild(tree objecttree.ObjectTree) error
- func (o *Object) Replaying() bool
- func (o *Object) Tree() objecttree.ObjectTree
- func (o *Object) TryClose(_ time.Duration) (bool, error)
- func (o *Object) Update(tree objecttree.ObjectTree) error
- type PlaintextSpec
- type TreeFunc
- type VersionAllocator
- type WriteResult
Constants ¶
This section is empty.
Variables ¶
var ErrClosed = errors.New("object: closed")
ErrClosed rejects writes on a closed (evicted) Object. Callers that resolved the Object before an eviction (e.g. the lazy schema-refresh Drop) retry with a fresh Store.Get.
var ErrEmptyPayload = errors.New("object: empty or non-object change payload")
ErrEmptyPayload is returned by Decode when the wire bytes are empty or the parsed top-level value is not an object.
var ErrTreeNotSet = errors.New("object: tree not set")
ErrTreeNotSet is retained for callers that historically distinguished "tree not yet bound" from other errors. The current constructor guarantees tree-on-construction, so this only surfaces if an Object is somehow zeroed out (defensive).
Functions ¶
This section is empty.
Types ¶
type AfterApply ¶
AfterApply fires after a successful ApplyChange — both the replay path and LocalWrite. Used by the space layer to drain parked changes whose missing shortIds may have just landed and to dispatch the change to subscribers.
The Object firing the hook is passed in so the space layer can reach the Controller directly without going through a cache lookup. That matters because afterApply runs from inside the LoadFunc on a fresh joiner (synctree's afterBuild → Rebuild → replayLocked → applyDecodedLocked → afterApply); any cache.Pick on the same id would block on the load channel that hasn't closed yet, producing a self-recursive deadlock.
res carries the per-record extras the apply path stamped beyond the input ops (handler-derived ops, _ver.id creation marker). Subscribers consume DerivedOps so a viewer can reconstruct a fresh record uniformly — no distinction between "user-supplied" and "auto" fields on the wire.
type AfterReplay ¶
AfterReplay fires once after replayLocked finishes a batch of inbound/replayed changes of which at least one applied — every per-change AfterApply of the batch has run. The space layer flushes work it coalesced per batch (object-stamp events for live queries). Runs under the same tree lock as the applies.
type ApplyGate ¶
type ApplyGate func(ctx context.Context, ch *crdt.Change, rawPayload []byte) (proceed bool, err error)
Object binds one any-sync object tree to one crdt.Controller.
Constructed via object.New, which builds the Object, runs the caller-supplied TreeFunc with the new Object as the synctree's update listener, and returns a fully-initialised Object with its tree bound. There is no intermediate state where an Object is returned without a tree — the listener wiring chicken/egg is resolved entirely inside New.
After construction the Object is live: LocalWrite produces new changes, and inbound sync calls fire Update/Rebuild which re-run the same replay path. Run ColdRestore once before publishing the Object to readers to drain any tree-changes the controller's MaxAddSeq watermark hasn't caught up to yet.
Not safe for concurrent LocalWrite — the apply path is serialised through a per-Object mutex. ApplyGate is the optional pre-apply hook the space layer wires to implement DataVersion-based deferral (the schema gate from docs/types-properties-proposal.md § "Detached-changes collection"). Called before each ApplyChange in the replay path; rawPayload is the decoded wire bytes (for parking).
- return (true, nil) → proceed with controller.ApplyChange.
- return (false, nil) → skip apply; the gate has parked the change for later drain.
- return (_, err) → the apply path logs and skips this row.
type Codec ¶
type Codec struct {
// contains filtered or unexported fields
}
Codec encodes/decodes the change payload. Pools its own arena and parser — both are cheap to reuse across many writes.
Not safe for concurrent use; the apply path is single-threaded per object and writes go through a per-object lock anyway.
func (*Codec) Decode ¶
Decode parses wire bytes into the caller-portion of a crdt.Change. The returned Change has Dataset, DataVersion, TraceIds, and Records populated; the CRDT envelope fields stay zero — the caller fills them in from the any-sync Change envelope before passing to Controller.ApplyChange.
Op.Payload values point into the codec's parser arena. They remain valid until the next Decode call. Callers that retain Records past the next Decode must clone the payload bytes.
type Config ¶
type Config struct {
SpaceId string
SignKey crypto.PrivKey
Controller *crdt.Controller
Allocator *VersionAllocator
Gate ApplyGate
AfterApply AfterApply
// AfterReplay runs once per replay batch after its applies (see
// the AfterReplay type). Optional.
AfterReplay AfterReplay
// WriteGate rejects user-authored DAG writes when it returns a
// non-nil error (read-only guest spaces). Optional — nil means
// writable.
WriteGate func() error
// PlaintextSpecs declares which tree-root ChangeTypes are plaintext
// object classes and which datasets they may carry. Optional — nil
// means every object writes encrypted changes.
PlaintextSpecs map[string]PlaintextSpec
// OnClose runs once when the Object is closed (eviction or cache
// shutdown), after the controller's own collection handles are
// released. The store hooks per-object resources the controller
// doesn't own (the `__history` collection handle). Optional.
OnClose func()
}
Config carries the dependencies object.New wires onto a new Object. Gate and AfterApply are optional (nil = disabled).
type Object ¶
type Object struct {
// contains filtered or unexported fields
}
func New ¶
New constructs a fully-initialised Object with its any-sync tree bound. The TreeFunc closure runs with the partially-built Object as its update listener; the returned tree is wired in atomically before New returns, so the Object is never observable to callers without a tree.
During TreeFunc execution any synctree listener callbacks (Update / Rebuild fired by BuildSyncTreeOrGetRemote's initial rebuild) see o.tree == nil — replayLocked handles that by stamping ObjectAuthor / Creator from the explicit tree arg it receives rather than from o.tree.
func (*Object) ApplyDecoded ¶
ApplyDecoded applies a pre-stamped Change to the controller — bypasses the gate, used by the drain path when re-applying a previously-parked change whose dependencies have now landed. Takes tree.Lock (the apply-serializing mutex) and runs applyDecodedLocked. Drain operations may wait briefly on inbound sync — that's fine, drain is off the critical path.
Rejects a closed Object like every other write entry point: post-close the controller's handles are released and collectionForWrite's open-by-name CREATES the collection if absent, so a drain landing on a just-closed (worse: just-purged) object would resurrect its `<objectId>_<dataset>` collection on disk. The drainer resolves objects via a fresh Store.Get per pass, so the retry lands on a live replacement.
func (*Object) Close ¶
Close detaches the tree listener, marks the Object closed, and closes the underlying tree. Blocks on tree.Lock — that's the same lock LocalWrite, the synchandler-driven Update, and the drain path all serialize on, so waiting here is the natural "let in-flight applies finish" barrier. Idempotent — second call is a no-op. ocache calls this on Remove and on cache shutdown.
The listener is nilled before the tree closes so any Update / Rebuild that races the close (any-sync's tree may still hold this *Object in memory via SyncAll iteration past eviction) no-ops instead of driving a stale apply against a freshly-loaded peer Object.
tree.Close drives any-sync's OnClose hook (objecttreebuilder.onClose → syncService.CloseReceiveQueue), which reaps the per-object multiqueue receive-queue goroutine. Without it that goroutine leaks for the process lifetime — eviction alone never reaches the synctree. tree.Close re-acquires tree.Lock internally, so it must run after the Unlock; the closed flag set under the lock guarantees exactly one caller reaches it.
The close path also releases the controller's per-object collection handles (and fires cfg.OnClose for store-owned extras) — each open any-store handle pins planner sketches and caches, so a TTL-evicted object must not keep its `<objectId>_<dataset>` handles pinned for the process lifetime. Safe here: applies serialize on tree.Lock and check o.closed, so once the flag is set under the lock no apply can touch the controller's collections; unlocked readers re-open by name (see Controller.CloseOwnedCollections).
func (*Object) ColdRestore ¶
ColdRestore replays everything from controller.MaxAddSeq forward. Run once at Open after SetTree. Acquires the tree lock — caller must NOT hold it.
func (*Object) Controller ¶
func (o *Object) Controller() *crdt.Controller
Controller returns the bound CRDT controller. Used for read paths that bypass the write-side locks (e.g. PropertiesAPI.Get hitting the controller's any-store collection directly).
func (*Object) InjectedSet ¶
InjectedSet applies an account-mirror materialization: ops on account-class fields (or per-key-scoped dynamic heads) written straight into the controller's materialised row — NO tree.AddContent, nothing enters THIS object's any-sync DAG. Unlike LocalSet the VersionId is CALLER-SUPPLIED: the tech-space carrier tree's orderId for the change that produced the value, so per-path gating replays the carrier's converged order exactly. Safe because account paths have exactly one writer per device (the mirror) sourcing one tech tree. Fires afterApply so Query/Subscribe and the applySeq feed see the change live, exactly like every other apply.
Caller passes a Change with Dataset + VersionId + Records (explicit ids, $set/$unset ops). ChangeId/AddSeq stay zero — the change never gets a DAG identity here; its provenance is the carrier change.
func (*Object) LocalSet ¶
LocalSet applies a device-local materialization: it writes only reserved local-namespace (crdt.LocalFieldPrefix) fields straight into the controller's materialised row — NO tree.AddContent, so nothing enters the any-sync DAG and nothing syncs to other devices. Each device computes its own value. The version is locally allocated (NextVersion of the field's current version), which is safe because no synced change ever writes a local-namespace path. Fires afterApply so Query/Subscribe see the change live, exactly like a synced write.
Caller passes a Change with Dataset + Records (explicit ids, $set/$unset ops on local-prefixed paths). VersionId/ChangeId/AddSeq are ignored on input — VersionId is assigned here; the change never gets a ChangeId.
func (*Object) LocalWrite ¶
LocalWrite applies a CRDT batch as a new tree change. Encodes the payload, signs and stores it via tree.AddContent, then runs the same per-change apply primitive the parked-replay drain uses (applyDecodedLocked) so there's a single apply path: any-sync generates the ChangeId/OrderId/AddSeq, the SDK stamps them onto the Change, and applyDecodedLocked lands it.
Returns VersionId (= any-sync OrderId), ChangeId, and the resolved record ids. Errors from encode/AddContent/apply propagate; partial state is impossible because AddContent and ApplyChange each run in their own atomic step.
Locking: tree.Lock is the single mutex guarding every write into the Controller. The sync-receiver path (synctree.AddRawChanges → Update/Rebuild → replayLocked) already runs under tree.Lock at the synctree layer; ColdRestore and ApplyDecoded acquire it themselves; LocalWrite acquires it here. tree.AddContent also requires the caller to hold tree.Lock — without it, any-sync logs "use tree when unlocked" at ERROR.
func (*Object) Rebuild ¶
func (o *Object) Rebuild(tree objecttree.ObjectTree) error
Rebuild fires when synctree had to rescan from a snapshot, also with the tree lock held by the caller.
func (*Object) Replaying ¶
Replaying reports whether the caller is inside a replayLocked batch (inbound sync, cold restore, re-index rebuild) as opposed to a single LocalWrite / drained change. Meaningful only from the AfterApply hook, which runs under the same lock.
func (*Object) Tree ¶
func (o *Object) Tree() objecttree.ObjectTree
Tree returns the bound tree, or nil if SetTree hasn't run.
func (*Object) TryClose ¶
TryClose is the non-blocking variant ocache GC uses. Returns (false, nil) when the tree is locked by an in-flight handler (LocalWrite, inbound synchandler, drain). ocache retries on the next tick. On success it closes the tree to reap the receive-queue goroutine — see Close for why the tree close runs unlocked and why collection handles are released here.
func (*Object) Update ¶
func (o *Object) Update(tree objecttree.ObjectTree) error
Update implements updatelistener.UpdateListener. Fired by synctree from inside AddRawChanges / buildSyncTree, with the tree lock already held — we must NOT re-lock here.
The tree must be opened with SetDeferredUpdater(true) — see spaceobjects/store.go:openTree. Without it, any-sync's default AddRawChangesWithUpdater order fires this listener BEFORE storage.AddAll, so replayLocked's IterateAfterAddSeq scan finds nothing new in storage and silently no-ops; tree heads advance while the controller stays out of sync.
type PlaintextSpec ¶
type PlaintextSpec struct {
Datasets map[string]struct{}
}
PlaintextSpec declares a plaintext (node-readable) object class: an object whose tree changes are written UNencrypted at the any-sync level (ShouldBeEncrypted:false → ReadKeyId=="" on the wire), so a reader without the space read key — e.g. a filenode-v2 broker — can materialize them. Field-level secrecy inside such changes is the dataset layer's job (an SDK-sealed field, see internal/payloads).
Datasets is the write allowlist. LocalWrite hard-errors on any other dataset (nothing leaks into the DAG); the inbound replay path tolerantly skips them (a peer can't smuggle rows into `objects` etc. through a plaintext tree).
type TreeFunc ¶
type TreeFunc func(listener updatelistener.UpdateListener) (objecttree.ObjectTree, error)
TreeFunc constructs the any-sync ObjectTree for the new Object, receiving the Object (as its UpdateListener) as input. Called exactly once from inside object.New; the returned tree is wired into the Object before New returns. Typical implementations call TreeBuilder.PutTree (creation/derivation path) and/or TreeBuilder.BuildTree (existing-tree path) with the listener passed straight through.
type VersionAllocator ¶
type VersionAllocator struct {
// contains filtered or unexported fields
}
VersionAllocator hands out monotonic lexids for the future device-scope write path — changes that don't propagate through any-sync at all (e.g. per-device UI state). Any-sync-routed writes (the only kind today) take their VersionId from any-sync's per-tree OrderId on the StorageChange returned by AddContent / emitted by GetAfterAddSeq; that lexid is owned and persisted by any-sync, which is why it survives process restart. Using a fresh-on-restart local allocator on those paths regressed LWW gating and silently dropped writes (any new write's versionId could be lower than fields stamped in a prior session).
Wiring is kept here so the device-scope path can land later without re-plumbing.
func NewVersionAllocator ¶
func NewVersionAllocator(last crdt.VersionId) *VersionAllocator
NewVersionAllocator initialises with `last` as the highest VersionId already in use. Empty string is the "no version yet" sentinel; the first Next() call returns the smallest lexid for the configured alphabet.
func (*VersionAllocator) Bump ¶
func (a *VersionAllocator) Bump(v crdt.VersionId)
Bump raises the watermark to at least v. Used during cold restore to advance the allocator past any VersionId already stamped on records before live traffic resumes.
func (*VersionAllocator) Last ¶
func (a *VersionAllocator) Last() crdt.VersionId
Last returns the most recently allocated VersionId without allocating a new one. Useful for persistence handshakes.
func (*VersionAllocator) Next ¶
func (a *VersionAllocator) Next() crdt.VersionId
Next allocates the next VersionId. Safe for concurrent use, though the apply path is single-threaded today.
type WriteResult ¶
type WriteResult struct {
VersionId crdt.VersionId
ChangeId string
RecordIds []string
Rejections []crdt.OpRejection
}
WriteResult bundles the identifiers a local write produces.
- VersionId is the peer-local lexid stamped onto the records.
- ChangeId is any-sync's content-addressable DAG change hash.
- RecordIds is the resolved id per record (post empty-id resolution); for empty-id records it equals base58(xxh3-64(ChangeId)) — the propId / shortId convention.
- Rejections lists per-op handler rejections — ops that the change carries but the handler refused to apply (kind mismatch, terminal status, immutable field, etc.). The change still committed with a fresh VersionId, but those ops did not land. Empty list means everything applied.
Source Files