reco

package module
v0.0.0-...-b883fbd Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2026 License: BSD-3-Clause Imports: 11 Imported by: 0

README

Go Reference

reco

EXPERIMENTAL RESEARCH PROTOTYPE — DO NOT USE. Nothing to see here; move along. This is unfinished research, not a supported library. Do not depend on it. APIs and behavior may change or disappear without notice. There are no compatibility or correctness guarantees.

Reco (reactive computation) explores typed, incremental dataflow graphs in Go. Declare mutable inputs and derived values; transactions update inputs, affected computations run, and subscribers receive changes. Persistent sets, maps, and typed records let incremental operators process small deltas without copying or rescanning entire collections. Local evaluation is synchronous; an experimental WebSocket transport connects explicitly bound graphs across processes.

See DESIGN.md for details and open questions.

Documentation

Overview

Package reco is an experimental research prototype for reactive computation.

EXPERIMENTAL: DO NOT USE

Nothing to see here; move along. This is unfinished research, not a supported library. Do not depend on it. APIs and behavior may change or disappear without notice. There are no compatibility or correctness guarantees.

Overview

Reco stands for reactive computation. It provides typed, in-process reactive dataflow graphs. Declare mutable leaves with Data, SetData, MultisetData, MapData, or StructData, and derived nodes with Func. Register an output with Graph.Register to register its dependencies too. Graph.Update applies a transaction, recomputes affected nodes in dependency order, and delivers subscription events after the graph has settled. Node definitions can be reused across independent Graph instances. Scope and In instantiate the same definitions independently within one graph; explicitly scoped inputs allow dependencies across scopes.

Each definition has a NodeClassName, supplied to its constructor and returned by Node.ClassName. Class names must be unique within a graph/scope, including its dependencies; they are distinct from collection keys and future instance keys.

Incremental operators

Package github.com/bradfitz/reco/nodes provides reusable incremental nodes, including set union and set-to-map computation. Packages can implement their own operators using Operator and Input. Allocate incremental caches inside the operator's per-instance compute factory. Consume collection changes with ChangesSince and publish one WithDelta batch per evaluation to preserve the incremental fast path. See the Operator example. Eval.ChangedInputs identifies changed dependency handles without scanning all declared inputs. The first evaluation reports all inputs; later evaluations report only value/result-status changes. It retains original declaration handles across scoping. Input reads the current settled value in O(1) expected time. Readiness and scope mapping do not rebuild input tables on point edits. Func still populates every field of its dependency struct; use Operator for wide, incrementally maintained computations. See the ChangedInputs example.

Published values must be immutable. SetSnapshot, MultisetSnapshot and MapSnapshot provide structurally shared storage, native Go iterators, and atomic delta application. Custom immutable types may implement ValueEqualer and VersionedValue. Multisets store positive int64 multiplicities, with signed atomic adjustments and checked underflow/overflow. Len counts distinct keys; All yields key/count pairs. nodes.CountBy and nodes.GroupCounts aggregate contributors, and nodes.Distinct suppresses count changes that do not cross zero. Indexes and counted classes can avoid expanding relationships into every node pair.

Demand-driven lifetimes

NewGraph keeps registered functions eager. NewGraphWithOptions with GraphOptions.DemandDriven enabled computes only subscribed dependency closures. Unsubscribe releases now-unused derived values and operator caches; ordinary authoritative leaves and registered definitions remain. Read temporarily activates a closure, returning an immutable snapshot without a lasting watch. SubscribeSnapshot, SubscribeMap, and SubscribeStruct atomically snapshot and subscribe to an output, so changes cannot fall between reading and watching. Reconnecting rebuilds released caches from current leaves. Stats exposes constant-time lifetime and evaluation counters. Reconfigure changes a function instance's inputs and compute factory inside a transaction while retaining its handle and subscribers. It requires a demand-driven graph and already-registered, lower-rank inputs. Newly needed inputs load before commit; abort releases temporary demand. Shared definitions in other graph instances remain unchanged.

BindDurable optionally binds a registered data node to an application-supplied loader. Its value is cached only while observed, and reloaded when demand returns. This works for scalars, sets, multisets, maps, records, and custom immutable values. Read/Subscribe return load errors without retaining partial demand. Set and collection mutation helpers reject these externally owned leaves. Commit backing storage inside Graph.Update, then use UpdateDurable to patch the resident cache; cold publications need neither loading nor computation. ReadCached probes residency without loading. Storage itself is application code: reco neither writes a database nor undoes an external commit. Loaders run synchronously under the graph lock and must not reenter it. Returned snapshots and loader closures can retain values independently; avoid capturing a full initial snapshot in a loader intended to save memory.

Typed records

StructSnapshot[T] uses an ordinary Go struct as a fixed, typed schema. Field handles provide typed edits and change inspection. SubscribeStruct atomically obtains an initial snapshot and watches settled field/collection changes. StructChanges.Then coalesces contiguous events without scanning unchanged collection contents. Use MapFieldChanges and SetFieldChanges on the composed batch rather than comparing its endpoint collections; MultisetFieldChanges provides the same touched-key behavior for counted fields. See the SubscribeStruct example for an incremental stream-encoding boundary.

Experimental remote watches

Package github.com/bradfitz/reco/metagraph connects explicit remote mirrors and authoritative exports using multiplexed bidirectional WebSockets. The core Graph remains local and synchronous. Remote-watch ownership and lifetime are currently application-managed, not automatically resolved through a Locator. See cmd/webdemo for a two-process example and the current limitations.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var ErrMultisetCount = errors.New("reco: multiset count out of range")

ErrMultisetCount reports a negative count or an int64 count overflow.

Functions

func ApplyMultisetDelta 

func ApplyMultisetDelta[K comparable](tx *Tx, node Node[MultisetSnapshot[K]], d MultisetDelta[K]) error

ApplyMultisetDelta applies one validated batch to the staged value. On error it leaves this call's staged value unchanged; return the error from Update to abort the entire transaction. Each call must leave valid counts, even if a subsequent call would undo it. Multiple calls retain the transaction's original delta base and normalize net-zero changes at commit.

func ApplySetDelta 

func ApplySetDelta[K comparable](tx *Tx, node Node[SetSnapshot[K]], d SetDelta[K])

ApplySetDelta applies d inside tx. Clear enumerates existing members so the prototype's point-change subscribers also receive every removal.

func ApplyStructDelta 

func ApplyStructDelta[T any](tx *Tx, node Node[StructSnapshot[T]], d StructDelta[T])

ApplyStructDelta applies an atomic field batch inside tx. Multiple calls compose against staged values and retain the transaction's original base.

func BindDurable 

func BindDurable[T any](g *Graph, node Node[T], load func() (T, error)) error

BindDurable makes a registered data node a cache of externally stored data. It requires a demand-driven graph and a node with no observers. Any current value is discarded. The binding belongs to this graph and scoped instance, not the reusable node definition.

load returns an immutable value and runs synchronously under the graph lock. It must not reenter the graph or acquire a lock held by a graph caller. Errors are returned by Read/Subscribe; failed activation releases all acquired demand and can be retried. The cached value is released with its last observer. Definitions, the loader, and snapshots retained by callers remain alive.

Set and collection mutation helpers reject bound nodes. To change one, commit its backing store and call UpdateDurable within the same Graph.Update callback. Only that callback may write the backing store while the graph is in use. This serializes reloads with durable commits and cache publication.

Example 
g := NewGraphWithOptions(GraphOptions{DemandDriven: true})
n := Data[string]("greeting")
g.Register(n)
stored := "hello" // Stand-in for a database record.
BindDurable(g, n, func() (string, error) { return stored, nil })
g.Update(func(tx *Tx) error {
	stored = "good morning" // Commit storage first; return errors before publication.
	UpdateDurable(tx, n, func(string) string { return stored })
	return nil
})
snapshot, _ := Read(g, n)
fmt.Println(snapshot.Value(), g.Stats().CachedDurableNodes)

Output:
good morning 0

func Deps 

func Deps[T any](v T) inlineDeps

Deps declares inline dependencies for Func.

func MapDelete 

func MapDelete[K comparable, V any](tx *Tx, node Node[MapSnapshot[K, V]], key K)

MapDelete removes one map entry.

func MapFieldChanges 

func MapFieldChanges[T any, K comparable, V any](field StructField[T, MapSnapshot[K, V]], c StructChanges[T]) iter.Seq[MapChange[K, V]]

MapFieldChanges yields a map field's net entry changes, even when c combines many events. It never re-diffs the endpoints. Entries have explicit before/ after presence bits, preserving nil and zero values. Order is unspecified.

func MapPut 

func MapPut[K comparable, V any](tx *Tx, node Node[MapSnapshot[K, V]], key K, value V)

MapPut upserts one map entry.

func MultisetFieldChanges 

func MultisetFieldChanges[T any, K comparable](field StructField[T, MultisetSnapshot[K]], c StructChanges[T]) iter.Seq[MultisetChange[K]]

MultisetFieldChanges yields a multiset field's net count changes, including after StructChanges.Then coalescing. It visits only changed keys, never re-diffing the endpoint collections. Zero counts denote absence.

func Reconfigure 

func Reconfigure[T any](tx *Tx, node Node[T], deps []Dependency, factory func() Compute[T]) (err error)

Reconfigure stages new inputs and a compute factory for a function instance, without replacing its handle or subscriptions. It requires a demand-driven graph. Inputs must already be registered at lower topological ranks than node; this restriction preserves ordering and rules out cycles without a graph scan. The reusable definition and other graph instances are not changed.

An active function acquires its new inputs now, returning loader errors before commit. The old inputs/output remain in use until the transaction succeeds. Aborting releases temporary demand. Commit replaces the operator cache and computes the new output atomically with other writes. An inactive function only changes its binding, without loading data. Configure each instance at most once per transaction. If committing external storage, call this first. Any returned error also aborts the transaction, even if the caller ignores it. Active bindings require initialized inputs (or inputs already staged for writing/reconfiguration in this transaction), so an uninitialized replacement cannot silently leave subscribers seeing the old computation's value.

func Set 

func Set[T any](tx *Tx, node Node[T], value T)

Set assigns a data node within a transaction. For externally owned leaves bound with BindDurable, use UpdateDurable; Set makes the transaction fail.

func SetDelete 

func SetDelete[K comparable](tx *Tx, node Node[SetSnapshot[K]], key K)

SetDelete removes one set key.

func SetFieldChanges 

func SetFieldChanges[T any, K comparable](field StructField[T, SetSnapshot[K]], c StructChanges[T]) iter.Seq[SetChange[K]]

SetFieldChanges yields a set field's net membership changes, even when c combines many events. It never re-diffs the endpoints. Order is unspecified.

func SetUpsert 

func SetUpsert[K comparable](tx *Tx, node Node[SetSnapshot[K]], key K)

SetUpsert adds one set key.

func SubscribeMap 

func SubscribeMap[K comparable, V any](g *Graph, node Node[MapSnapshot[K, V]], opts SubscribeOptions, fn func(MapEvent[K, V])) (Snapshot[MapSnapshot[K, V]], SubscriptionHandle, error)

SubscribeMap atomically returns the current snapshot and subscribes fn to later transaction-local map mutations. The atomic setup prevents updates from being lost between taking the initial snapshot and starting the watch.

func SubscribeSnapshot 

func SubscribeSnapshot[T any](g *Graph, node Node[T], opts SubscribeOptions, fn func(Event[T])) (Snapshot[T], SubscriptionHandle, error)

SubscribeSnapshot atomically reads node and installs a watch for subsequent changes. Unlike a separate Read followed by Subscribe, no update can fall in between. The callback may run as soon as this function releases the graph lock, including before the caller has processed the returned snapshot. Callbacks run under that lock and must not reenter the graph or do network I/O.

func SubscribeStruct 

func SubscribeStruct[T any](g *Graph, node Node[StructSnapshot[T]], opts SubscribeOptions, fn func(StructEvent[T])) (Snapshot[StructSnapshot[T]], SubscriptionHandle, error)

SubscribeStruct atomically returns the current snapshot and watches later changes. An invalid initial snapshot means the node has not produced a value; use the first event as the initial state in that case. Events may start before this function returns, so initialize callback state before calling it.

Callbacks run synchronously under the graph lock after propagation. Capture or accumulate the immutable changes there; encode/compress/write outside the callback. Changes.Then supports coalescing without scanning full snapshots, but callers must bound pending state and decide when to disconnect/resync a slow consumer. Options have the same current limitations as Subscribe. Snapshot/Event still do not expose Result error/partial status; validity alone does not establish application-level completeness or readiness.

Example 
package main

import (
	"fmt"
	"slices"

	"github.com/bradfitz/reco"
	"github.com/bradfitz/reco/nodes"
)

func main() {
	// A miniature control-server meta record. The actual application can use
	// NodeID/PeerInfo and add DNS, filter, policy, and other typed fields.
	type meta struct {
		Domain string
		Peers  reco.MapSnapshot[int, string]
	}
	domain := reco.Data[string]("domain")
	peers := reco.MapData[int, string]("peers")
	mrm := nodes.Struct[meta]("mapResponseMeta", struct {
		Domain reco.Node[string]
		Peers  reco.Node[reco.MapSnapshot[int, string]]
	}{domain, peers})
	domainField := reco.Field[meta, string]("Domain")
	peerField := reco.Field[meta, reco.MapSnapshot[int, string]]("Peers")
	g := reco.NewGraph()
	if err := g.Register(mrm); err != nil {
		panic(err)
	}
	update := func(fn func(*reco.Tx)) {
		if err := g.Update(func(tx *reco.Tx) error { fn(tx); return nil }); err != nil {
			panic(err)
		}
	}
	update(func(tx *reco.Tx) {
		reco.Set(tx, domain, "example.net")
		reco.MapPut(tx, peers, 1, "original")
		reco.MapPut(tx, peers, 2, "leaving")
	})

	// Callbacks only accumulate immutable semantic changes. A real stream
	// writer would synchronize access to pending and bound its retained size.
	// Serialization, compression, and network writes happen outside callbacks.
	var pending reco.StructChanges[meta]
	initial, sub, err := reco.SubscribeStruct(g, mrm, reco.SubscribeOptions{}, func(ev reco.StructEvent[meta]) {
		var err error
		pending, err = pending.Then(ev.Changes)
		if err != nil {
			panic(err)
		} // A production stream would terminate/resync.
	})
	if err != nil {
		panic(err)
	}
	defer sub.Unsubscribe()
	fmt.Println("initial:", initial.Value().Value().Domain, initial.Value().Value().Peers.Len(), "peers")

	// Updates can arrive while the initial snapshot is being sent.
	update(func(tx *reco.Tx) { reco.MapPut(tx, peers, 1, "intermediate"); reco.MapPut(tx, peers, 3, "joined") })
	update(func(tx *reco.Tx) {
		reco.MapPut(tx, peers, 1, "final")
		reco.MapDelete(tx, peers, 2)
		reco.Set(tx, domain, "new.example.net")
	})

	if _, current, changed := domainField.Change(pending); changed {
		fmt.Println("domain:", current)
	}
	// Sort only the changed peers, not the full live map. Even though the
	// writer coalesced events, no deep endpoint-snapshot comparison is needed.
	changes := slices.Collect(reco.MapFieldChanges(peerField, pending))
	slices.SortFunc(changes, func(a, b reco.MapChange[int, string]) int { return a.Key - b.Key })
	for _, change := range changes {
		if change.AfterValid {
			fmt.Println("upsert:", change.Key, change.After)
		} else {
			fmt.Println("remove:", change.Key)
		}
	}
}

Output:
initial: example.net 2 peers
domain: new.example.net
upsert: 1 final
remove: 2
upsert: 3 joined

func UpdateDurable 

func UpdateDurable[T any](tx *Tx, node Node[T], update func(T) T)

UpdateDurable publishes an already-persisted mutation to a durable leaf's cache. If cold, it does nothing, including not calling update: a later observer reloads the latest stored value. If resident, update receives the current immutable value (including earlier publications in this transaction). Return a delta-based snapshot to preserve incremental downstream computation.

Commit the backing store inside the same Graph.Update callback, before calling UpdateDurable, and return its error without publishing if the commit fails. Do all fallible preparation before committing; graph rollback cannot undo an external commit. This function never writes storage itself.

Types

type Compute 

type Compute[T any] func(Eval) Result[T]

Compute evaluates an operator using its declared inputs. Its closure may retain instance-local incremental caches, but must not mutate published values. The result must depend only on the inputs, not on the cache's history.

type Dep 

type Dep[T any] struct {
	Value_     T
	Version_   Version
	Valid_     bool
	Err_       error
	IsPartial_ bool
}

Dep is the typed value wrapper used by typed-struct dependency inputs.

func Input 

func Input[T any](eval Eval, node Node[T]) Dep[T]

Input returns the current value, version, and result status of a declared dependency during a computation. It panics for a zero or undeclared node or outside an evaluation. It does not acquire a graph lock or create a watch. Unlike Changes(), collection ChangesSince(previous) safely handles both unchanged inputs and replacements since the operator last evaluated them.

func (Dep[T]) Err 

func (d Dep[T]) Err() error

Err returns the dependency error, if any.

func (Dep[T]) IsPartial 

func (d Dep[T]) IsPartial() bool

IsPartial reports whether the dependency value is partial.

func (Dep[T]) Valid 

func (d Dep[T]) Valid() bool

Valid reports whether the dependency is available.

func (Dep[T]) Value 

func (d Dep[T]) Value() T

Value returns the dependency value.

func (Dep[T]) Version 

func (d Dep[T]) Version() Version

Version returns the dependency version.

type Dependency 

type Dependency interface {
	// contains filtered or unexported methods
}

Dependency is a node that an operator declares as an input. Every Node[T] implements Dependency, allowing heterogeneous or variable-length input lists. Create nodes with Data, Func, or Operator; the runtime owns their handles.

type Eval 

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

Eval represents one synchronous node evaluation. Use Input to read its declared dependencies; ChangedInputs identifies which dependencies changed. It is not a cancellation or deadline context. Do not retain an Eval after the computation returns.

func (Eval) ChangedInputs 

func (e Eval) ChangedInputs() iter.Seq[Dependency]

ChangedInputs yields declared input handles whose value or result status changed since this computation's previous invocation. The first invocation of a compute instance yields all inputs, including after demand eviction or Reconfigure. Errors returned by Compute do not defer this boundary: retain unapplied work if recovery needs changes from earlier failed evaluations.

Handles are Node[T] values of the declared types, so callers can type assert or look them up in a map keyed by Dependency. Scoping preserves the handles from the original declaration, just as Input does. Repeated declarations of the same node yield it once; distinct declared handles that resolve to the same scoped instance are each yielded. Order is unspecified. All values read with Input are settled for this transaction, whether changed or not.

The iterator is reusable within this evaluation, supports early termination, and visits only changed handles. Do not retain it or Eval after computation returns. It panics outside an evaluation. Func can use it too, but populating Func's dependency struct still visits all of that struct's fields.

Example 
package main

import (
	"fmt"

	"github.com/bradfitz/reco"
)

func main() {
	a, b, c := reco.Data[int]("a"), reco.Data[int]("b"), reco.Data[int]("c")
	sum := reco.Operator("sum", []reco.Dependency{a, b, c}, func() reco.Compute[int] {
		previous := make(map[reco.Node[int]]int)
		total := 0
		return func(e reco.Eval) reco.Result[int] {
			// All three inputs on first evaluation; afterwards only the inputs
			// that changed. Duplicate declarations would be reported once.
			for input := range e.ChangedInputs() {
				n := input.(reco.Node[int]) // This operator declares only int inputs.
				current := reco.Input(e, n).Value()
				total += current - previous[n]
				previous[n] = current
			}
			return reco.OK(total)
		}
	})
	g := reco.NewGraph()
	if err := g.Register(sum); err != nil {
		panic(err)
	}
	if err := g.Update(func(tx *reco.Tx) error {
		reco.Set(tx, a, 10)
		reco.Set(tx, b, 20)
		reco.Set(tx, c, 30)
		return nil
	}); err != nil {
		panic(err)
	}
	value, _ := reco.Read(g, sum)
	fmt.Println(value.Value())
	if err := g.Update(func(tx *reco.Tx) error { reco.Set(tx, b, 25); return nil }); err != nil {
		panic(err)
	}
	value, _ = reco.Read(g, sum)
	fmt.Println(value.Value())
}

Output:
60
65

type Event 

type Event[T any] struct {
	Previous Snapshot[T]
	Current  Snapshot[T]
	Version  Version
}

Event is delivered to subscribers when a node value changes.

type Graph 

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

Graph is an independent reactive dataflow graph instance.

func NewGraph 

func NewGraph() *Graph

NewGraph creates an empty, eager graph instance. See NewGraphWithOptions for demand-driven evaluation and automatic cache release.

func NewGraphWithOptions 

func NewGraphWithOptions(opts GraphOptions) *Graph

NewGraphWithOptions creates a graph with the specified evaluation lifetime.

func (*Graph) Register 

func (g *Graph) Register(nodes ...any) error

Register atomically adds definitions and dependencies, validating class name uniqueness within each Scope and DAG structure. Work is proportional to newly registered definitions and their edges, not the existing graph's size.

func (*Graph) Stats 

func (g *Graph) Stats() GraphStats

Stats returns a constant-time snapshot of graph counters.

func (*Graph) Update 

func (g *Graph) Update(fn func(*Tx) error) error

Update runs one serialized transaction and then recomputes affected active functions to a fixed point before delivering subscription events. In eager graphs all registered functions are active; demand-driven graphs activate only subscribed dependency closures.

type GraphOptions 

type GraphOptions struct {
	// DemandDriven evaluates only functions reachable from subscriptions.
	// Read evaluates on demand without retaining a watch. When the last
	// observer goes away, derived values and operator caches are released.
	// Registered definitions and ordinary data values remain available.
	// BindDurable permits reloadable data values to be released too.
	DemandDriven bool
}

GraphOptions configures evaluation lifetime.

type GraphStats 

type GraphStats struct {
	Nodes, Functions, ActiveFunctions, CachedFunctions, Subscriptions int
	DurableNodes, CachedDurableNodes                                  int
	Evaluations                                                       uint64
	// InputChecks counts dependency readiness checks, including first-demand
	// initialization. InputReads counts values requested by computations (also
	// including Func's struct adapter). They measure work, not elapsed time.
	InputChecks, InputReads uint64
}

GraphStats reports graph lifetime and work counters. Registered definitions and ordinary data leaves outlive subscriptions. CachedFunctions counts retained operator states. DurableNodes counts loader bindings and CachedDurableNodes counts their resident values. Demand-driven graphs release both types of cache with last demand, without unregistering definitions.

type MapChange 

type MapChange[K comparable, V any] struct {
	Key         K
	Before      V
	BeforeValid bool
	After       V
	AfterValid  bool
}

MapChange describes one key changed by a map transaction. Before and After are accompanied by validity bits so zero values remain unambiguous.

type MapDelta 

type MapDelta[K comparable, V any] struct {
	Clear  bool
	Remove []K
	Put    []MapEntry[K, V]
}

MapDelta describes one atomic clear, remove, then put batch. Put wins over removal; repeated Put keys use the last entry in the slice. A nil value is distinct from removal. Clear plus Put replaces the entire map.

type MapEntry 

type MapEntry[K comparable, V any] struct {
	Key   K
	Value V
}

MapEntry is a whole key/value assignment, including a nil value if V allows it.

type MapEvent 

type MapEvent[K comparable, V any] struct {
	Previous MapSnapshot[K, V]
	Current  MapSnapshot[K, V]
	Changes  []MapChange[K, V]
	Version  Version
}

MapEvent is delivered after a transaction changes a map node. Changes lists only changed keys. Consecutive snapshots use their recorded delta; arbitrary snapshot replacements are reconciled against the previous value.

type MapSnapshot 

type MapSnapshot[K comparable, V any] struct {
	// contains filtered or unexported fields
}

MapSnapshot is a stable, structurally shared snapshot of a map node value. Point updates copy only the HAMT path to the changed key. The zero value is an empty map; use WithDelta to construct or update one. Entry values use ValueEqualer when implemented, otherwise Go equality for comparable values and reflect.DeepEqual for other values. Pointer values are compared by identity, not by the contents of the objects they point to.

func (MapSnapshot[K, V]) All 

func (s MapSnapshot[K, V]) All() iter.Seq2[K, V]

All returns an iterator over the snapshot's key/value pairs in unspecified order. It visits entries lazily without copying the backing map and stops when the range loop breaks. Each iteration starts afresh over this snapshot, even if the graph has since changed. A zero snapshot yields no entries.

Example 
package main

import (
	"fmt"
	"slices"

	"github.com/bradfitz/reco"
)

func main() {
	lengths := reco.MapData[string, int]("lengths")
	g := reco.NewGraph()
	if err := g.Register(lengths); err != nil {
		panic(err)
	}
	if err := g.Update(func(tx *reco.Tx) error {
		reco.MapPut(tx, lengths, "cedar", 5)
		reco.MapPut(tx, lengths, "oak", 3)
		return nil
	}); err != nil {
		panic(err)
	}
	snapshot, err := reco.Read(g, lengths)
	if err != nil {
		panic(err)
	}
	var entries []string
	for word, length := range snapshot.Value().All() {
		entries = append(entries, fmt.Sprintf("%s:%d", word, length))
	}
	slices.Sort(entries) // Iteration order is unspecified.
	fmt.Println(entries)
}

Output:
[cedar:5 oak:3]

func (MapSnapshot[K, V]) Changes 

func (s MapSnapshot[K, V]) Changes() []MapChange[K, V]

Changes returns a copy of the net point mutations that produced this snapshot. It is not relative to an arbitrary reader's previous snapshot; use ChangesSince when consuming updates, including unchanged inputs and snapshot replacements.

func (MapSnapshot[K, V]) ChangesSince 

func (s MapSnapshot[K, V]) ChangesSince(previous MapSnapshot[K, V]) iter.Seq[MapChange[K, V]]

ChangesSince iterates net entry changes from previous to s. It uses a recorded delta when its base matches previous, otherwise compares the snapshots (O(N)). Identical roots yield nothing. Each key appears at most once, in unspecified order, with whole before/after values and explicit presence bits.

func (MapSnapshot[K, V]) Get 

func (s MapSnapshot[K, V]) Get(k K) (V, bool)

Get returns a value by key.

func (MapSnapshot[K, V]) Len 

func (s MapSnapshot[K, V]) Len() int

Len returns the number of map entries.

func (MapSnapshot[K, V]) Range 

func (s MapSnapshot[K, V]) Range(fn func(K, V) bool)

Range calls fn for each entry in unspecified order until fn returns false.

func (MapSnapshot[K, V]) RecoValueEqual 

func (s MapSnapshot[K, V]) RecoValueEqual(other any) bool

RecoValueEqual implements ValueEqualer using persistent-root identity.

func (MapSnapshot[K, V]) RecoWithVersion 

func (s MapSnapshot[K, V]) RecoWithVersion(version Version) any

RecoWithVersion implements VersionedValue without mutating this snapshot.

func (MapSnapshot[K, V]) Version 

func (s MapSnapshot[K, V]) Version() Version

Version returns the graph-local version at which this snapshot was observed.

func (MapSnapshot[K, V]) WithDelta 

func (s MapSnapshot[K, V]) WithDelta(d MapDelta[K, V]) MapSnapshot[K, V]

WithDelta returns a new snapshot after applying d, without modifying s. It preserves HAMT structural sharing, normalizes repeated touched keys, and returns s for a net-zero change. Changed snapshots have Version zero until observed by a graph. Values must be treated as immutable by callers. Prefer one batch per operator evaluation; see SetSnapshot.WithDelta.

Example 
package main

import (
	"fmt"

	"github.com/bradfitz/reco"
)

func main() {
	var names reco.MapSnapshot[int, string] // An empty map, no graph required.
	names = names.WithDelta(reco.MapDelta[int, string]{
		Put: []reco.MapEntry[int, string]{{Key: 1, Value: "old"}},
	})
	replacement := names.WithDelta(reco.MapDelta[int, string]{
		Clear: true,
		Put:   []reco.MapEntry[int, string]{{Key: 2, Value: "new"}},
	})
	fmt.Println(names.Get(1))
	fmt.Println(replacement.Get(2))
}

Output:
old true
new true

type MultisetChange 

type MultisetChange[K comparable] struct {
	Key           K
	Before, After int64
}

MultisetChange is a net multiplicity change. Zero denotes absence. Before and After are nonnegative; After-Before is the signed adjustment.

type MultisetDelta 

type MultisetDelta[K comparable] struct {
	Clear  bool
	Remove []K
	Put    []MultisetEntry[K]
	Adjust map[K]int64
}

MultisetDelta is an atomic clear, remove, put, then adjust batch. Remove deletes all occurrences. Repeated Put keys use the last assignment. Adjust adds one signed net change per key; it does not assign a count. Every Put count and every final adjusted count must be in [0, math.MaxInt64]. Clear plus Put replaces the multiset. Input slices/maps are not retained.

Additive adjustments are not idempotent: replaying +1 adds another occurrence. Remote delivery still needs base/version checks; this is not a CRDT counter.

type MultisetEntry 

type MultisetEntry[K comparable] struct {
	Key   K
	Count int64
}

MultisetEntry assigns a key's multiplicity. Count must be nonnegative; zero removes the key. Counts are per key, not a bounded collection total.

type MultisetSnapshot 

type MultisetSnapshot[K comparable] struct {
	// contains filtered or unexported fields
}

MultisetSnapshot is an immutable counted set, backed by a persistent map. The zero value is empty. Only positive counts are stored; removing one occurrence does not remove a key until its last occurrence disappears. Keys use Go equality and must be reflexive (no NaNs), as for SetSnapshot. Interface keys must be dynamically comparable.

Example 
package main

import (
	"fmt"

	"github.com/bradfitz/reco"
)

func main() {
	var counts reco.MultisetSnapshot[string]
	counts, err := counts.WithDelta(reco.MultisetDelta[string]{
		Put: []reco.MultisetEntry[string]{{Key: "web", Count: 2}},
	})
	if err != nil {
		panic(err)
	}
	counts, err = counts.WithDelta(reco.MultisetDelta[string]{Adjust: map[string]int64{"web": -1}})
	if err != nil {
		panic(err)
	}
	fmt.Println(counts.Len(), counts.Count("web"))
	// Decrementing below zero fails atomically; the previous value is returned.
	unchanged, err := counts.WithDelta(reco.MultisetDelta[string]{Adjust: map[string]int64{"web": -2}})
	fmt.Println(err != nil, unchanged.Count("web"))
}

Output:
1 1
true 1

func (MultisetSnapshot[K]) All 

func (s MultisetSnapshot[K]) All() iter.Seq2[K, int64]

All lazily iterates distinct key/count pairs, as Range does.

func (MultisetSnapshot[K]) Changes 

func (s MultisetSnapshot[K]) Changes() []MultisetChange[K]

Changes returns a copy of this snapshot's last net count changes, not changes relative to an arbitrary reader. No-op edits retain old metadata, so prefer ChangesSince when consuming updates.

func (MultisetSnapshot[K]) ChangesSince 

func (s MultisetSnapshot[K]) ChangesSince(previous MultisetSnapshot[K]) iter.Seq[MultisetChange[K]]

ChangesSince yields one net change per touched key in unspecified order. Identical roots yield nothing, adjacent deltas take O(delta) work plus map traversal, and skipped bases/replacements may require O(N) reconciliation.

func (MultisetSnapshot[K]) Count 

func (s MultisetSnapshot[K]) Count(k K) int64

Count returns k's multiplicity, or zero if absent.

func (MultisetSnapshot[K]) Counts 

func (s MultisetSnapshot[K]) Counts() MapSnapshot[K, int64]

Counts returns an immutable map view without copying. Updating the returned map does not modify the multiset. There is no unchecked map-to-multiset cast.

func (MultisetSnapshot[K]) Len 

func (s MultisetSnapshot[K]) Len() int

Len returns the number of DISTINCT keys, not the sum of their counts.

func (MultisetSnapshot[K]) Range 

func (s MultisetSnapshot[K]) Range(fn func(K, int64) bool)

Range visits each distinct key and its positive count in unspecified order, stopping when fn returns false. It does not repeat a key Count times.

func (MultisetSnapshot[K]) RecoValueEqual 

func (s MultisetSnapshot[K]) RecoValueEqual(other any) bool

RecoValueEqual implements ValueEqualer using persistent-root identity.

func (MultisetSnapshot[K]) RecoWithVersion 

func (s MultisetSnapshot[K]) RecoWithVersion(v Version) any

RecoWithVersion implements VersionedValue without modifying this snapshot.

func (MultisetSnapshot[K]) Version 

func (s MultisetSnapshot[K]) Version() Version

Version returns the graph-local observation version.

func (MultisetSnapshot[K]) WithDelta 

func (s MultisetSnapshot[K]) WithDelta(d MultisetDelta[K]) (MultisetSnapshot[K], error)

WithDelta applies d atomically. An invalid count returns s and an error wrapping ErrMultisetCount, without applying ANY part of d. Underflow is an error, not a clamped deletion. Net-zero updates reuse the original root. Point edits do touched-key work plus HAMT path copying; Clear enumerates existing keys to report removals. Changed snapshots have Version zero.

type Node 

type Node[T any] struct {
	// contains filtered or unexported fields
}

Node is a typed handle to a value in a graph.

func Data 

func Data[T any](className NodeClassName) Node[T]

Data creates a mutable data node.

func Func 

func Func[DepsT any, Out any](className NodeClassName, deps any, compute func(Eval, DepsT) Result[Out]) Node[Out]

Func creates a function node using either a typed binding struct or an inline dependency declaration created with Deps.

func In 

func In[T any](scope *Scope, node Node[T]) Node[T]

In returns the instance of node in scope, recursively instantiating its unscoped dependencies. Already-scoped nodes are preserved, allowing explicit dependencies between scopes in the same graph. ClassName is unchanged. In is safe to call concurrently. Nil scope and zero node handles panic.

Example 
package main

import (
	"fmt"

	"github.com/bradfitz/reco"
)

func main() {
	input := reco.Data[int]("input")
	doubled := reco.Func("doubled", reco.Deps(struct{ N reco.Node[int] }{input}),
		func(_ reco.Eval, in struct{ N int }) reco.Result[int] {
			return reco.OK(2 * in.N)
		})
	var left, right reco.Scope
	a, b := reco.In(&left, doubled), reco.In(&right, doubled)
	g := reco.NewGraphWithOptions(reco.GraphOptions{DemandDriven: true})
	if err := g.Register(a, b); err != nil {
		panic(err)
	}
	if err := g.Update(func(tx *reco.Tx) error {
		reco.Set(tx, reco.In(&left, input), 3)
		reco.Set(tx, reco.In(&right, input), 7)
		return nil
	}); err != nil {
		panic(err)
	}
	// A one-shot read computes current state without keeping it active.
	av, err := reco.Read(g, a)
	if err != nil {
		panic(err)
	}
	bv, err := reco.Read(g, b)
	if err != nil {
		panic(err)
	}
	fmt.Println(av.Value(), bv.Value())
	fmt.Println("active functions:", g.Stats().ActiveFunctions)
}

Output:
6 14
active functions: 0

func MapData 

func MapData[K comparable, V any](className NodeClassName) Node[MapSnapshot[K, V]]

MapData creates a mutable map data node.

func MultisetData 

func MultisetData[K comparable](name NodeClassName) Node[MultisetSnapshot[K]]

MultisetData declares a mutable counted-set leaf. Use ApplyMultisetDelta or Set with a validated snapshot. Ordinary SubscribeSnapshot observes its changes.

func Operator 

func Operator[T any](className NodeClassName, deps []Dependency, newCompute func() Compute[T]) Node[T]

Operator declares a reusable incremental node. newCompute is called lazily, per graph/node instance, to create its compute function and state. In demand- driven graphs, the state is discarded when last demand disappears and the factory runs again on reactivation. Caches must be rebuildable from inputs. Allocate mutable caches INSIDE newCompute, not in the shared definition. A factory must return a non-nil function. It can run concurrently with factories/computations in other graphs; calls within one graph are serialized.

Compute reads declared inputs with Input. The runtime waits until they are initialized, schedules affected nodes in dependency order, and delivers the settled results. A compute function must not call Read, Update, or Subscribe on its graph, perform external side effects, or retain its Eval.

For collections, use ChangesSince to consume input deltas and WithDelta to publish a new immutable output. A normal stateless computation can use Func. For many inputs, Eval.ChangedInputs avoids inspecting unchanged dependencies. It reports changes since the previous evaluation, even if that evaluation returned an error; operators must retain any unapplied work needed to recover.

Example 
package main

import (
	"fmt"
	"strings"

	"github.com/bradfitz/reco"
)

func main() {
	words := reco.SetData[string]("words")
	// This is an external-package implementation of a key-preserving map.
	uppercase := reco.Operator("uppercase", []reco.Dependency{words}, func() reco.Compute[reco.MapSnapshot[string, string]] {
		// These caches belong to one graph, not to the reusable definition.
		var previous reco.SetSnapshot[string]
		var result reco.MapSnapshot[string, string]
		return func(eval reco.Eval) reco.Result[reco.MapSnapshot[string, string]] {
			current := reco.Input(eval, words).Value()
			var delta reco.MapDelta[string, string]
			for c := range current.ChangesSince(previous) {
				if c.Present {
					// Only newly added words are converted.
					delta.Put = append(delta.Put, reco.MapEntry[string, string]{Key: c.Key, Value: strings.ToUpper(c.Key)})
				} else {
					delta.Remove = append(delta.Remove, c.Key)
				}
			}
			previous = current
			result = result.WithDelta(delta) // One atomic, structurally shared batch.
			return reco.OK(result)
		}
	})
	g := reco.NewGraph()
	if err := g.Register(uppercase); err != nil {
		panic(err)
	}
	if err := g.Update(func(tx *reco.Tx) error {
		reco.ApplySetDelta(tx, words, reco.SetDelta[string]{Add: []string{"oak", "elm"}})
		return nil
	}); err != nil {
		panic(err)
	}
	first, err := reco.Read(g, uppercase)
	if err != nil {
		panic(err)
	}
	if err := g.Update(func(tx *reco.Tx) error {
		reco.ApplySetDelta(tx, words, reco.SetDelta[string]{Remove: []string{"elm"}, Add: []string{"pine"}})
		return nil
	}); err != nil {
		panic(err)
	}
	current, err := reco.Read(g, uppercase)
	if err != nil {
		panic(err)
	}
	fmt.Println(current.Value().Get("oak"))
	fmt.Println(current.Value().Get("pine"))
	fmt.Println(first.Value().Get("elm")) // Previously published values stay intact.
}

Output:
OAK true
PINE true
ELM true

func SetData 

func SetData[K comparable](className NodeClassName) Node[SetSnapshot[K]]

SetData creates a mutable set data node.

func StructData 

func StructData[T any](className NodeClassName) Node[StructSnapshot[T]]

StructData declares a mutable record data node and validates T's schema.

func (Node[T]) ClassName 

func (n Node[T]) ClassName() NodeClassName

ClassName returns the name of the node's definition, or "" for a zero handle.

func (Node[T]) Valid 

func (n Node[T]) Valid() bool

Valid reports whether this handle references a node definition.

type NodeClassName 

type NodeClassName string

NodeClassName names a reusable node definition (a node class). Constructors such as Data, Func, and Operator take a class name; Node.ClassName returns it.

A name must be nonempty. Names are exact, case-sensitive strings: reco does not trim whitespace, normalize text, or interpret separators or punctuation. There is no required namespace or path syntax. The zero value is invalid for a definition, but is returned by ClassName on a zero Node handle.

Within a Graph and Scope, all definitions, including dependencies, must have distinct names regardless of their value types or constructors. Unscoped definitions share the graph's default namespace. Registering the same definition again is allowed; separately constructed definitions with the same name are an error, not aliases. Definitions and names can be reused across independent graphs or instantiated with In in multiple scopes; names are not globally unique.

A class name is not a collection key or a key selecting an instance of a class. Scope provides local instances; generic keyed resolution and remote addressing are not yet implemented; this type does not define a distributed address or versioning scheme.

type Result 

type Result[T any] struct {
	Value     T
	Err       error
	IsPartial bool
}

Result is the value produced by a function node.

A partial result is still a value and may be propagated. Err is deliberately part of Result rather than a second return value so compute functions keep all failure state explicit in the dataflow model.

func OK 

func OK[T any](v T) Result[T]

OK returns a complete successful Result.

func Partial 

func Partial[T any](v T, err error) Result[T]

Partial returns a partial Result with an error explaining what is incomplete.

type Scope 

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

Scope gives reusable definitions independent instances within one Graph. It is a local namespace, not a separate graph, transaction boundary, or security boundary. A Scope must not be copied after first use. Its zero value is ready for use; retain it to obtain the same instances on subsequent calls.

type SetChange 

type SetChange[K comparable] struct {
	Key     K
	Present bool
}

SetChange describes one key changed by a set transaction.

type SetDelta 

type SetDelta[K comparable] struct {
	Clear  bool
	Remove []K
	Add    []K
}

SetDelta is an atomic clear, remove, then add batch. Add wins when a key is also removed. The enclosing transaction may batch changes to other nodes.

type SetSnapshot 

type SetSnapshot[K comparable] struct {
	// contains filtered or unexported fields
}

SetSnapshot is a stable, structurally shared snapshot of a set node value. The zero value is an empty set; use WithDelta to construct or update one.

func (SetSnapshot[K]) All 

func (s SetSnapshot[K]) All() iter.Seq[K]

All returns an iterator over the snapshot's keys in unspecified order. It visits keys lazily without copying the backing set and stops when the range loop breaks. Each iteration starts afresh over this snapshot, even if the graph has since changed. A zero snapshot yields no keys.

Example 
package main

import (
	"fmt"
	"slices"

	"github.com/bradfitz/reco"
)

func main() {
	words := reco.SetData[string]("words")
	g := reco.NewGraph()
	if err := g.Register(words); err != nil {
		panic(err)
	}
	if err := g.Update(func(tx *reco.Tx) error {
		reco.ApplySetDelta(tx, words, reco.SetDelta[string]{Add: []string{"cedar", "birch"}})
		return nil
	}); err != nil {
		panic(err)
	}
	snapshot, err := reco.Read(g, words)
	if err != nil {
		panic(err)
	}
	var names []string
	for word := range snapshot.Value().All() {
		names = append(names, word)
	}
	slices.Sort(names) // Iteration order is unspecified.
	fmt.Println(names)
}

Output:
[birch cedar]

func (SetSnapshot[K]) Changes 

func (s SetSnapshot[K]) Changes() []SetChange[K]

Changes returns a copy of the net point mutations that produced this snapshot. It is not relative to an arbitrary reader's previous snapshot; use ChangesSince when consuming updates, including unchanged inputs and snapshot replacements.

func (SetSnapshot[K]) ChangesSince 

func (s SetSnapshot[K]) ChangesSince(previous SetSnapshot[K]) iter.Seq[SetChange[K]]

ChangesSince iterates net membership changes from previous to s. Consecutive snapshots use their recorded delta in O(delta) work; identical backing roots yield nothing. Initial snapshots, skipped versions, and arbitrary replacements fall back to comparing memberships, which can take O(N). Versions alone are not used to establish lineage, since snapshots can come from different graphs. Each key is yielded at most once, in unspecified order.

func (SetSnapshot[K]) Contains 

func (s SetSnapshot[K]) Contains(k K) bool

Contains reports whether k is present.

func (SetSnapshot[K]) Len 

func (s SetSnapshot[K]) Len() int

Len returns the number of set entries.

func (SetSnapshot[K]) Range 

func (s SetSnapshot[K]) Range(fn func(K) bool)

Range calls fn for each key in unspecified order until fn returns false.

func (SetSnapshot[K]) RecoValueEqual 

func (s SetSnapshot[K]) RecoValueEqual(other any) bool

RecoValueEqual implements ValueEqualer using persistent-root identity.

func (SetSnapshot[K]) RecoWithVersion 

func (s SetSnapshot[K]) RecoWithVersion(version Version) any

RecoWithVersion implements VersionedValue without mutating this snapshot.

func (SetSnapshot[K]) Version 

func (s SetSnapshot[K]) Version() Version

Version returns the graph-local version at which this snapshot was observed.

func (SetSnapshot[K]) WithDelta 

func (s SetSnapshot[K]) WithDelta(d SetDelta[K]) SetSnapshot[K]

WithDelta returns a new snapshot after applying d, without modifying s or any published value. Point edits copy only their immutable HAMT paths. Clear enumerates removals for the point-change event API. Net-zero edits return s. A changed result has Version zero until observed by a graph, and its Changes describe the net changes from s. A no-op returns s, including its old Changes; use ChangesSince(s) to distinguish this case. Prefer one batch per operator evaluation to preserve the O(delta) path through ChangesSince and SubscribeMap.

type Snapshot 

type Snapshot[T any] struct {
	// contains filtered or unexported fields
}

Snapshot is the immutable view delivered through subscription events.

For scalar values, Value returns the value itself. Collection snapshots can use richer concrete types while still carrying the same graph version.

func Read 

func Read[T any](g *Graph, node Node[T]) (Snapshot[T], error)

Read returns the latest value of a registered node. In demand-driven mode it temporarily observes its dependency closure and releases that demand before returning; the returned immutable snapshot remains valid.

func ReadCached 

func ReadCached[T any](g *Graph, node Node[T]) (Snapshot[T], error)

ReadCached returns a node's resident snapshot without loading, computing, or adding demand. An invalid snapshot means the node is cold or uninitialized. It is useful for point reads backed by a store lookup when a collection is cold. Returned snapshots retain their values independently of graph demand.

func (Snapshot[T]) Valid 

func (s Snapshot[T]) Valid() bool

Valid reports whether the snapshot contains a value.

func (Snapshot[T]) Value 

func (s Snapshot[T]) Value() T

Value returns the snapshot value.

func (Snapshot[T]) Version 

func (s Snapshot[T]) Version() Version

Version returns the graph-local version at which this snapshot was observed.

type StructChanges 

type StructChanges[T any] struct {
	// contains filtered or unexported fields
}

StructChanges is an immutable net change batch between two record snapshots. It retains only its endpoints and touched-key changes, not an event history. A zero value is an empty accumulator. Then composes contiguous batches without re-diffing collection snapshots; external synchronization is needed only when assigning a shared accumulator, not for reading a published batch.

func (StructChanges[T]) After 

func (c StructChanges[T]) After() StructSnapshot[T]

After returns the final snapshot, retaining the composed changes from Before.

func (StructChanges[T]) Before 

func (c StructChanges[T]) Before() StructSnapshot[T]

Before returns the initial snapshot of the batch.

func (StructChanges[T]) ChangeCount 

func (c StructChanges[T]) ChangeCount() int

ChangeCount counts scalar field replacements and map/set entry changes. It inspects only the fixed field list, not pending collection entries. This can help bound a stream accumulator; it is not a byte-size or memory bound.

func (StructChanges[T]) Fields 

func (c StructChanges[T]) Fields() iter.Seq[string]

Fields yields changed field names in schema order. It inspects only the fixed field list, never collection contents. The iterator is reusable.

func (StructChanges[T]) Len 

func (c StructChanges[T]) Len() int

Len returns the number of fields with nonempty net changes.

func (StructChanges[T]) Then 

func (c StructChanges[T]) Then(next StructChanges[T]) (StructChanges[T], error)

Then composes a subsequent batch. It rejects a gap or out-of-order batch: next.Before must match this batch's actual After root and observed version, not a reconstructed equal value. Versions alone cannot establish lineage across graphs. A zero accumulator accepts any first batch. Work is proportional to the fixed field count and incoming touched keys plus persistent-map path copying.

type StructDelta 

type StructDelta[T any] []StructEdit[T]

StructDelta is an ordered atomic batch of field edits. Repeated edits see earlier values; the last Set wins. An empty batch does nothing. Fields cannot be added or deleted. Edits and referenced values must not be mutated while the batch is being applied.

type StructEdit 

type StructEdit[T any] struct {
	// contains filtered or unexported fields
}

StructEdit is an opaque typed field edit, created by StructField.Set or Update. Its zero value is invalid.

type StructEvent 

type StructEvent[T any] struct {
	Previous StructSnapshot[T]
	Current  StructSnapshot[T]
	Changes  StructChanges[T]
	Version  Version
}

StructEvent describes one settled record change. Changes includes typed field replacements and net map/set mutations. Copying the event is cheap and safe; all published snapshots and change batches are immutable.

type StructField 

type StructField[T, V any] struct {
	// contains filtered or unexported fields
}

StructField is a typed handle to a declared field of T. Its zero value is invalid. Create one with Field, normally once alongside node definitions.

func Field 

func Field[T, V any](name string) StructField[T, V]

Field finds an exported, non-embedded field by its Go name. It panics if T is not a supported struct, the field is missing, or its type is not exactly V. Validation happens at declaration time; edits and reads remain typed.

func (StructField[T, V]) Change 

func (f StructField[T, V]) Change(c StructChanges[T]) (before, after V, changed bool)

Change returns this field's before/after values and whether it changed. When unchanged, both returned values are zero. For collection entry changes, use MapFieldChanges or SetFieldChanges instead of diffing these snapshots.

func (StructField[T, V]) Get 

func (f StructField[T, V]) Get(s StructSnapshot[T]) V

Get returns the field's typed value.

func (StructField[T, V]) Name 

func (f StructField[T, V]) Name() string

Name returns the field's Go name.

func (StructField[T, V]) Set 

func (f StructField[T, V]) Set(value V) StructEdit[T]

Set replaces this field, including with a nil, zero, false, or empty value.

func (StructField[T, V]) Update 

func (f StructField[T, V]) Update(fn func(V) V) StructEdit[T]

Update transforms this field, seeing earlier edits in the same atomic delta. fn must be pure and must not mutate its input. For collection fields, return current.WithDelta(batch) to preserve delta-sized work.

type StructSnapshot 

type StructSnapshot[T any] struct {
	// contains filtered or unexported fields
}

StructSnapshot is an immutable record whose schema is the Go struct T. T must be a struct with exported, non-embedded fields. Value returns a typed shallow copy; referenced objects must remain immutable, as with other reco values. A zero snapshot contains the zero value of T.

Fields use ValueEqualer when available, otherwise ordinary scalar equality. MapSnapshot and SetSnapshot fields preserve their incremental changes. Other fields are whole-value replacements, not recursively patched Go objects. Record overhead is proportional to the fixed field count; scalar comparison still follows each value's equality contract. Unchanged map/set contents are never scanned. Use Field to declare typed edits and inspect changes, and SubscribeStruct to obtain an initial snapshot plus future changes.

func NewStruct 

func NewStruct[T any](value T) StructSnapshot[T]

NewStruct wraps value without copying its referenced storage. It validates the schema but does not enumerate collection contents or calculate changes.

func (StructSnapshot[T]) ChangesSince 

func (s StructSnapshot[T]) ChangesSince(previous StructSnapshot[T]) StructChanges[T]

ChangesSince describes the net field/entry changes from previous to s. Adjacent retained collection deltas avoid scanning unchanged contents. If collection versions were skipped or replaced without lineage, reconciliation may enumerate them. Capture each event and use StructChanges.Then to coalesce efficiently, rather than discarding intermediate snapshots and diffing later.

func (StructSnapshot[T]) RecoValueEqual 

func (s StructSnapshot[T]) RecoValueEqual(other any) bool

RecoValueEqual compares field values, respecting their ValueEqualer hooks. Collection fields compare persistent roots, not backing-store contents.

func (StructSnapshot[T]) RecoWithVersion 

func (s StructSnapshot[T]) RecoWithVersion(v Version) any

RecoWithVersion implements VersionedValue without changing field versions.

func (StructSnapshot[T]) Value 

func (s StructSnapshot[T]) Value() (value T)

Value returns the typed record. Do not mutate objects referenced by it.

func (StructSnapshot[T]) Version 

func (s StructSnapshot[T]) Version() Version

Version returns the graph-local version at which this record was observed.

func (StructSnapshot[T]) WithDelta 

func (s StructSnapshot[T]) WithDelta(d StructDelta[T]) StructSnapshot[T]

WithDelta applies all edits without publishing intermediate states. It coalesces nested collection changes, preserving their original delta base. Net-zero edits reuse the previous storage. Whole collection replacements without retained delta lineage may require enumeration; point edits do not.

Example 
package main

import (
	"fmt"

	"github.com/bradfitz/reco"
)

func main() {
	type settings struct {
		Enabled bool
		Names   reco.SetSnapshot[string]
	}
	enabled := reco.Field[settings, bool]("Enabled")
	names := reco.Field[settings, reco.SetSnapshot[string]]("Names")
	before := reco.NewStruct(settings{Enabled: true})
	after := before.WithDelta(reco.StructDelta[settings]{
		enabled.Set(false), // Explicit false, not an omitted change.
		names.Update(func(s reco.SetSnapshot[string]) reco.SetSnapshot[string] {
			return s.WithDelta(reco.SetDelta[string]{Add: []string{"elm"}})
		}),
	})
	changes := after.ChangesSince(before)
	if old, value, changed := enabled.Change(changes); changed {
		fmt.Println("enabled:", old, "->", value)
	}
	for change := range reco.SetFieldChanges(names, changes) {
		fmt.Println("member:", change.Key, change.Present)
	}
	fmt.Println("old snapshot:", before.Value().Enabled, before.Value().Names.Len())
}

Output:
enabled: true -> false
member: elm true
old snapshot: true 0

type SubscribeOptions 

type SubscribeOptions struct {
	FixedPointOnly bool
	Coalesce       bool
}

SubscribeOptions controls subscription delivery.

type SubscriptionHandle 

type SubscriptionHandle interface {
	Unsubscribe() error
}

SubscriptionHandle cancels a subscription.

func Subscribe 

func Subscribe[T any](g *Graph, node Node[T], opts SubscribeOptions, fn func(Event[T])) (SubscriptionHandle, error)

Subscribe registers a callback for node value changes, without an initial event. In demand-driven mode it keeps the dependency closure active until Unsubscribe. Callbacks run under the graph lock and must not reenter it.

type Tx 

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

Tx is a serialized per-graph transaction.

type ValueEqualer 

type ValueEqualer interface {
	RecoValueEqual(other any) bool
}

ValueEqualer optionally replaces reflect.DeepEqual for a custom value type. RecoValueEqual must be a pure equivalence test, handle a different concrete type by returning false, and compare semantic values rather than versions or delta metadata. Immutable values can compare persistent-root identities. Implement this on the published value type (not only on its pointer type if the graph stores values). Scalar values without this hook use reflect.DeepEqual; map entries use Go equality when comparable, otherwise reflect.DeepEqual.

type Version 

type Version uint64

Version is a monotonic graph-local version.

type VersionedValue 

type VersionedValue interface {
	RecoWithVersion(Version) any
}

VersionedValue optionally stamps an observed graph version into a custom value. RecoWithVersion must return the SAME concrete type and must not mutate the receiver or previously published snapshots. This hook is independent of ValueEqualer; most custom values need neither hook. Snapshot.Version remains available even when the value does not implement VersionedValue.

Source Files

View all Source files

Directories

Path Synopsis
cmd
webdemo command
Command webdemo runs an in-memory reco playground with optional peer watches.
 Command webdemo runs an in-memory reco playground with optional peer watches.
Package metagraph is an EXPERIMENTAL, unstable prototype of multiplexed cross-process reco watches.
 Package metagraph is an EXPERIMENTAL, unstable prototype of multiplexed cross-process reco watches.
Package nodes provides reusable incremental node definitions for reco graphs.
 Package nodes provides reusable incremental node definitions for reco graphs.
Package recotestcontrol is an experimental, reco-backed Tailscale test control server.
 Package recotestcontrol is an experimental, reco-backed Tailscale test control server.

Jump to

Keyboard shortcuts

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