diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 0b91f71..17d95d4 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -39,6 +39,9 @@ type Tool struct { Name string Description string Input json.RawMessage + // Effect is what the tool does, reads, acts or dangerous (novox/hq ADR 0387 §1), where the runtime knows it: its own + // tools say it; a bundle's tool is marked by its manifest. + Effect string // Run calls the tool for caller: the bus user the call's subject named, or "" when it named none. A // caller is handed to the bundle in the call's `_meta` (MetaCaller), never among its arguments. Run func(caller string, args json.RawMessage) (json.RawMessage, error) diff --git a/node-tools/internal/ledger/read_test.go b/node-tools/internal/ledger/read_test.go index 17584db..2539c63 100644 --- a/node-tools/internal/ledger/read_test.go +++ b/node-tools/internal/ledger/read_test.go @@ -8,8 +8,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscalls" ) -// 1966/AC15 and the other reads of a ledger: get, query (ADR 0597 §10; design 25 § 12.5). -func TestReadsOfALedger(t *testing.T) { +// 1966/AC15, the other reads of a ledger (get, query) and its repairs (ADR 0597 §8, §10; design 25 § 12.5). +func TestReadsAndRepairsOfALedger(t *testing.T) { cases := []struct { name string run func(t *testing.T, l *Ledger, f fakes) @@ -57,6 +57,59 @@ func TestReadsOfALedger(t *testing.T) { t.Fatalf("an undeclared index: %v", err) } }}, + {"a lost state rebuilds from the history, an expired one from its snapshot, and lag says what is behind", func(t *testing.T, l *Ledger, f fakes) { + applied(t, l, "decision/1", "title", `{"title":"One"}`, owner) + applied(t, l, "decision/1", "note", `{"says":"kept"}`, owner) + want, _, _ := f.state.Get("decision.1") + f.state.kv = nil + lag, _ := l.Lag(buscalls.LedgerRequest{}) + if len(lag.Instances) != 1 || lag.Instances[0].History != 2 || lag.Instances[0].State != 0 || lag.Instances[0].Snapshot != 2 { + t.Fatalf("lag says %+v", lag) + } + if got, err := l.Rebuild(buscalls.RebuildRequest{}); err != nil || got.Written != 1 { + t.Fatalf("rebuilt %+v, %v", got, err) + } + if got, _, _ := f.state.Get("decision.1"); string(record(t, got).State) != string(record(t, want).State) { + t.Fatalf("rebuilt from the history %s, want %s", got, want) + } + f.state.kv = nil + f.history.Expire() + if got, err := l.Rebuild(buscalls.RebuildRequest{Instance: "decision/1", From: buscalls.RebuildFromSnapshot}); err != nil || got.Written != 1 { + t.Fatalf("rebuilt %+v, %v", got, err) + } + if got, _, _ := f.state.Get("decision.1"); string(got) != string(want) { + t.Fatalf("rebuilt from the snapshot %s, want %s", got, want) + } + }}, + {"reindex builds a declared index from the states, and lag says an index behind until it does", func(t *testing.T, l *Ledger, f fakes) { + applied(t, l, "decision/1", "title", `{"title":"One ledger"}`, owner) + applied(t, l, "decision/2", "note", `{"says":"untitled"}`, owner) + _ = f.index.Put("stale.decision.9", []byte(`{}`)) + if lag, _ := l.Lag(buscalls.LedgerRequest{}); len(lag.Indexes) != 1 || lag.Indexes[0].Behind != 1 { + t.Fatalf("lag says %+v", lag) + } + if got, err := l.Reindex(buscalls.ReindexRequest{Index: "by-title"}); err != nil || got.Written != 1 { + t.Fatalf("reindexed %+v, %v", got, err) + } + entries, _ := f.index.Entries("", 0) + if len(entries) != 1 || entries[0].Key != "One_ledger.decision.1" || string(entries[0].Value) != `{"instance":"decision/1","sequence":1}` { + t.Fatalf("the index holds %+v", entries) + } + if lag, _ := l.Lag(buscalls.LedgerRequest{}); len(lag.Indexes) != 0 || len(lag.Instances) != 0 { + t.Fatalf("lag after reindex says %+v", lag) + } + }}, + {"snapshot rewrites a snapshot behind its state", func(t *testing.T, l *Ledger, f fakes) { + applied(t, l, "decision/1", "title", `{"title":"One"}`, owner) + f.snapshots.kv = nil + if got, err := l.Snapshot(buscalls.InstanceRequest{}); err != nil || got.Written != 1 { + t.Fatalf("snapshot %+v, %v", got, err) + } + state, _, _ := f.state.Get("decision.1") + if snap, _, _ := f.snapshots.Get("decision.1"); string(snap) != string(state) { + t.Fatalf("the snapshot %s, the state %s", snap, state) + } + }}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { diff --git a/node-tools/internal/ledger/repair.go b/node-tools/internal/ledger/repair.go new file mode 100644 index 0000000..e0d2803 --- /dev/null +++ b/node-tools/internal/ledger/repair.go @@ -0,0 +1,315 @@ +package ledger + +import ( + "encoding/json" + "fmt" + "regexp" + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// The ledger's repairs (ADR 0597 §8, §10; design 25 § 12.5; issue 1966 T4): rebuild an instance's state from its history +// or its snapshot, rewrite a snapshot behind its state, build a declared index again, and say what is behind its history. +// The tool runner serves them as tools of every module owning a ledger, its own ledgers only (the runtime's repairs.go). + +// instances is every instance the ledger knows, by key: those its state holds and those its history names. +func (l *Ledger) instances() ([]string, error) { + seen := map[string]bool{} + keys, err := l.State.Keys() + if err != nil { + return nil, err + } + for _, k := range keys { + seen[k] = true + } + kept, err := l.History.Read(">", 1, 0) + if err != nil { + return nil, err + } + ops, err := operationsOf(kept) + if err != nil { + return nil, err + } + for _, o := range ops { + if k, err := KeyOf(o.Instance); err == nil { + seen[k] = true + } + } + out := make([]string, 0, len(seen)) + for k := range seen { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +// keysOf is the one instance a request names, or every instance when it names none. +func (l *Ledger) keysOf(instance string) ([]string, error) { + if instance == "" { + return l.instances() + } + k, err := KeyOf(instance) + if err != nil { + return nil, err + } + return []string{k}, nil +} + +// Rebuild derives the state of an instance (every instance when it names none) again, from its whole history or from its +// snapshot and the operations after it, and writes it, and its snapshot when that is behind. A derivation further back +// than the state holds is not written: the state never goes back (ADR 0597 §5). +func (l *Ledger) Rebuild(r buscalls.RebuildRequest) (buscalls.Repaired, error) { + from := r.From + if from == "" { + from = buscalls.RebuildFromHistory + } + if from != buscalls.RebuildFromHistory && from != buscalls.RebuildFromSnapshot { + return buscalls.Repaired{}, fmt.Errorf("a rebuild is from %q or %q, not %q", buscalls.RebuildFromHistory, buscalls.RebuildFromSnapshot, from) + } + keys, err := l.keysOf(r.Instance) + if err != nil { + return buscalls.Repaired{}, err + } + var out buscalls.Repaired + for _, key := range keys { + base := Record{} + if from == buscalls.RebuildFromSnapshot { + if base, _, err = l.readSnapshot(key); err != nil { + return out, err + } + } + derived, err := l.replay(key, base, 0) + if err != nil { + return out, err + } + held, rev, err := l.readState(key) + if err != nil { + return out, err + } + if derived.Sequence < held.Sequence { + continue + } + now, _, err := l.settle(key, derived, rev) + if err != nil { + return out, err + } + if err := l.writeSnapshot(key, now); err != nil { + return out, err + } + out.Written++ + } + return out, nil +} + +// Snapshot rewrites the snapshot of an instance (every instance when it names none) that is behind its state. +func (l *Ledger) Snapshot(r buscalls.InstanceRequest) (buscalls.Repaired, error) { + keys, err := l.keysOf(r.Instance) + if err != nil { + return buscalls.Repaired{}, err + } + var out buscalls.Repaired + for _, key := range keys { + held, _, err := l.readState(key) + if err != nil { + return out, err + } + snap, at, err := l.readSnapshot(key) + if err != nil { + return out, err + } + if held.Sequence == 0 || (at != 0 && snap.Sequence >= held.Sequence) { + continue + } + if err := l.writeSnapshot(key, held); err != nil { + return out, err + } + out.Written++ + } + return out, nil +} + +// unsafe is what a key part cannot hold. +var unsafe = regexp.MustCompile(`[^-_A-Za-z0-9]`) + +// IndexEntryOf is an instance's entry in a declared index, as reindex writes it: its key the values of the fields the +// index is on, in their order, then the instance's key, each part with what a key cannot hold written as `_`; its value +// the instance and the sequence of the state it was taken from. False when the state holds none of those fields. +func IndexEntryOf(on []string, key string, rec Record) (string, []byte, bool) { + var state map[string]any + if json.Unmarshal(rec.State, &state) != nil || state == nil { + return "", nil, false + } + parts := make([]string, 0, len(on)+1) + found := false + for _, f := range on { + v, has := state[f] + if !has { + parts = append(parts, "_") + continue + } + found = true + s, ok := v.(string) + if !ok { + raw, _ := json.Marshal(v) + s = string(raw) + } + if s = unsafe.ReplaceAllString(s, "_"); s == "" { + s = "_" + } + parts = append(parts, s) + } + if !found { + return "", nil, false + } + value, _ := json.Marshal(map[string]any{"instance": InstanceOf(key), "sequence": rec.Sequence}) + return strings.Join(append(parts, key), "."), value, true +} + +// declared is the fields an index is on and the types declaring it. +func (l *Ledger) declared(index string) ([]string, map[string]bool, error) { + types := map[string]bool{} + var on []string + for _, t := range l.Types { + for _, ix := range t.Indexes { + if ix.Name == index { + types[t.Name] = true + on = ix.On + } + } + } + if len(types) == 0 { + return nil, nil, fmt.Errorf("%s's ledger %s declares no index %q", l.Module, l.Name, index) + } + return on, types, nil +} + +// wanted is every entry an index should hold, from the states of the types declaring it. +func (l *Ledger) wanted(index string) (map[string][]byte, map[string]uint64, error) { + on, types, err := l.declared(index) + if err != nil { + return nil, nil, err + } + keys, err := l.instances() + if err != nil { + return nil, nil, err + } + want, seqs := map[string][]byte{}, map[string]uint64{} + for _, key := range keys { + name, _, _ := strings.Cut(key, ".") + if !types[name] { + continue + } + held, _, err := l.readState(key) + if err != nil { + return nil, nil, err + } + if k, v, ok := IndexEntryOf(on, key, held); ok { + want[k], seqs[k] = v, held.Sequence + } + } + return want, seqs, nil +} + +// Reindex builds a declared index again from the ledger's states, derived from its history: every entry it should hold +// written, every other removed. +func (l *Ledger) Reindex(r buscalls.ReindexRequest) (buscalls.Repaired, error) { + ix, err := l.index(r.Index) + if err != nil { + return buscalls.Repaired{}, err + } + want, _, err := l.wanted(r.Index) + if err != nil { + return buscalls.Repaired{}, err + } + held, err := ix.Entries("", 0) + if err != nil { + return buscalls.Repaired{}, err + } + var out buscalls.Repaired + for _, e := range held { + if _, keep := want[e.Key]; !keep { + if err := ix.Delete(e.Key); err != nil { + return out, err + } + } + } + keys := make([]string, 0, len(want)) + for k := range want { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + if err := ix.Put(k, want[k]); err != nil { + return out, err + } + out.Written++ + } + return out, nil +} + +// Lag is every instance whose state or snapshot is behind its history, and how many entries each declared index lacks +// or holds from an older state; nothing when nothing is behind. +func (l *Ledger) Lag(buscalls.LedgerRequest) (buscalls.Lagged, error) { + keys, err := l.instances() + if err != nil { + return buscalls.Lagged{}, err + } + var out buscalls.Lagged + for _, key := range keys { + last, err := l.History.Last(key) + if err != nil { + return out, err + } + held, _, err := l.readState(key) + if err != nil { + return out, err + } + snap, _, err := l.readSnapshot(key) + if err != nil { + return out, err + } + if held.Sequence < last || snap.Sequence < last { + out.Instances = append(out.Instances, buscalls.InstanceLag{Instance: InstanceOf(key), History: last, + State: held.Sequence, Snapshot: snap.Sequence}) + } + } + names := make([]string, 0, len(l.Indexes)) + for name := range l.Indexes { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + if _, _, err := l.declared(name); err != nil { + // A bucket bound for an index no type declares has nothing it should hold. + continue + } + want, seqs, err := l.wanted(name) + if err != nil { + return out, err + } + held, err := l.Indexes[name].Entries("", 0) + if err != nil { + return out, err + } + at := map[string]uint64{} + for _, e := range held { + var v struct { + Sequence uint64 `json:"sequence"` + } + _ = json.Unmarshal(e.Value, &v) + at[e.Key] = v.Sequence + } + behind := uint64(0) + for k := range want { + if s, has := at[k]; !has || s < seqs[k] { + behind++ + } + } + if behind > 0 { + out.Indexes = append(out.Indexes, buscalls.IndexLag{Index: name, Behind: behind}) + } + } + return out, nil +} diff --git a/node-tools/internal/runtime/ledgers.go b/node-tools/internal/runtime/ledgers.go new file mode 100644 index 0000000..3b5b053 --- /dev/null +++ b/node-tools/internal/runtime/ledgers.go @@ -0,0 +1,146 @@ +package runtime + +import ( + "encoding/json" + "errors" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + "github.com/novox/mesh-tools/node-tools/internal/launch" + "github.com/novox/mesh-tools/node-tools/internal/ledger" +) + +// A module's ledgers as the tool runner serves them (novox/hq ADR 0597, design 25 § 12.4, § 12.5; issue 1966): opened +// for the module whose bundle or whose repair tool asks, from its own bindings alone, with its types and the connection +// its history is published on. + +// LedgerTypes is where the runner reads a ledger's types, by module and local name. Where it reads them from on the +// live mesh is not decided (issue 1966: the binding carries no types, and no record names a source); until it is, +// the runner refuses to open a ledger and says why, and a test gives the types. +type LedgerTypes func(module, local string) ([]catalogue.LedgerType, error) + +// errTypesUndecided is the refusal of a ledger whose types the runner has no source for. +func errTypesUndecided(module, local string) error { + return fmt.Errorf("the tool runner cannot open %s's ledger %q: it reads a ledger's types from nowhere yet, since its "+ + "binding carries none and no record names their source (issue 1966)", module, local) +} + +// ledgerSource is how the runner opens a module's ledger: its types, and the connection its history and snapshots are +// published on (the module's ledger credential, ADR 0597 §9). +type ledgerSource struct { + types LedgerTypes + writer func(module string) (*bus.Conn, error) +} + +// ledger opens one of a module's ledgers by its local name. +func (c *consumers) ledger(module, local string) (*ledger.Ledger, error) { + b, err := c.conn.Binding(module, local, catalogue.StorageLedger) + if err != nil { + return nil, err + } + types := c.ledgers.types + if types == nil { + return nil, errTypesUndecided(module, local) + } + ts, err := types(module, local) + if err != nil { + return nil, err + } + var writer *bus.Conn + if c.ledgers.writer != nil { + if writer, err = c.ledgers.writer(module); err != nil { + return nil, err + } + } + history, state, snapshots := c.conn.LedgerStores(b, writer) + return &ledger.Ledger{Module: module, Name: local, Types: ts, History: history, State: state, Snapshots: snapshots, + Indexes: c.conn.LedgerIndexes(b), Logf: c.logf}, nil +} + +// hasLedger is whether a module's membership binds a ledger. +func hasLedger(conn *bus.Conn, module string) bool { + for _, b := range conn.Bindings(module) { + if b.Storage == catalogue.StorageLedger { + return true + } + } + return false +} + +// The repair tools the runner serves for every module owning a ledger (design 25 § 12.5), named so no bundle's tool is +// taken for one. +const ( + RepairRebuild = "ledger_rebuild" + RepairReindex = "ledger_reindex" + RepairSnapshot = "ledger_snapshot" + RepairLag = "ledger_lag" +) + +// The effects a tool carries (novox/hq ADR 0387 §1). +const ( + EffectReads = "reads" + EffectActs = "acts" +) + +// repairTools are a module's ledger repairs as tools, each limited to its own ledgers: the request's ledger is a local +// name of the module whose tool is called. +func (c *consumers) repairTools(module string) []launch.Tool { + schema := func(props string, required ...string) json.RawMessage { + req, _ := json.Marshal(append([]string{"ledger"}, required...)) + return json.RawMessage(`{"type":"object","properties":{"ledger":{"type":"string","description":"the ledger's local name"}` + + props + `},"required":` + string(req) + `}`) + } + instance := `,"instance":{"type":"string","description":"one instance, such as decision/0290; every instance when empty"}` + run := func(do func(l *ledger.Ledger, args json.RawMessage) (any, error)) func(string, json.RawMessage) (json.RawMessage, error) { + return func(_ string, args json.RawMessage) (json.RawMessage, error) { + var named struct { + Ledger string `json:"ledger"` + } + if err := json.Unmarshal(args, &named); err != nil || named.Ledger == "" { + return nil, errors.New("a repair names the ledger, by its local name") + } + l, err := c.ledger(module, named.Ledger) + if err != nil { + return nil, err + } + answer, err := do(l, args) + if err != nil { + return nil, err + } + return json.Marshal(answer) + } + } + return []launch.Tool{ + {Name: RepairRebuild, Effect: EffectActs, Input: schema(instance + `,"from":{"type":"string","enum":["history","snapshot"],"description":"derive from the whole history (the default) or from the snapshot and the operations after it"}`), + Description: "Derive a ledger instance's state again from its history or its snapshot, and write it and its snapshot (novox/hq ADR 0597 §10).", + Run: run(func(l *ledger.Ledger, args json.RawMessage) (any, error) { + var r buscalls.RebuildRequest + _ = json.Unmarshal(args, &r) + return l.Rebuild(r) + })}, + {Name: RepairReindex, Effect: EffectActs, Input: schema(`,"index":{"type":"string","description":"the declared index"}`, "index"), + Description: "Build a ledger's declared index again from its states, derived from its history (novox/hq ADR 0597 §4, §10).", + Run: run(func(l *ledger.Ledger, args json.RawMessage) (any, error) { + var r buscalls.ReindexRequest + _ = json.Unmarshal(args, &r) + return l.Reindex(r) + })}, + {Name: RepairSnapshot, Effect: EffectActs, Input: schema(instance), + Description: "Rewrite the snapshot of a ledger instance that is behind its state (novox/hq ADR 0597 §8, §10).", + Run: run(func(l *ledger.Ledger, args json.RawMessage) (any, error) { + var r buscalls.InstanceRequest + _ = json.Unmarshal(args, &r) + return l.Snapshot(r) + })}, + {Name: RepairLag, Effect: EffectReads, Input: schema(""), + Description: "Say every ledger instance whose state or snapshot is behind its history, and each declared index behind it (novox/hq ADR 0597 §10).", + Run: run(func(l *ledger.Ledger, args json.RawMessage) (any, error) { + var r buscalls.LedgerRequest + _ = json.Unmarshal(args, &r) + return l.Lag(r) + })}, + } +} diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go new file mode 100644 index 0000000..15ba59f --- /dev/null +++ b/node-tools/internal/runtime/ledgers_test.go @@ -0,0 +1,74 @@ +package runtime + +import ( + "encoding/json" + "errors" + "sort" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/membership" + + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// 1966/AC16: given a module with a ledger binding, when the runner lists its tools, then the four repair tools are +// among them, marked acts or reads; a module without one has none, and a binding written later brings them. +func TestTheLedgerRepairsAreToolsOfEveryLedgerOwner(t *testing.T) { + mesh := mt.New(t) + owner := mt.MembershipOf("owner", "anchor", true, nil) + owner.Bindings = []membership.BusBinding{{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, + History: "LEDGER_owner_records", Subject: "mesh.ledger.owner.records", State: "owner_records", + Snapshot: "mesh.ledger-snapshot.owner.records"}} + plain := mt.MembershipOf("plain", "anchor", true, nil) + mesh.Issue(t, owner) + mesh.Issue(t, plain) + nodeTools := connect(t, "node-tools", "anchor") + stop, err := Run(nodeTools, []Served{{"owner", []string{mt.Fixture("caller-echo.py")}}, + {"plain", []string{mt.Fixture("caller-echo.py")}}}, nil, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + defer stop() + asker := connect(t, "console", "workstation") + listed := func(module string) (string, error) { + answered, err := asker.Ask(module+"."+ToolsVerb, map[string]any{}, "") + if err != nil { + return "", err + } + var got ToolsAnswer + if err := json.Unmarshal(answered.Result, &got); err != nil { + return "", err + } + var names []string + for _, tool := range got.Tools { + names = append(names, tool.Name+":"+tool.Effect) + } + sort.Strings(names) + return strings.Join(names, " "), nil + } + for _, c := range []struct{ name, module, want string }{ + {"a ledger owner lists the four repairs", "owner", "ledger_lag:reads ledger_rebuild:acts ledger_reindex:acts ledger_snapshot:acts who:"}, + {"a module with no ledger has none", "plain", "who:"}, + } { + t.Run(c.name, func(t *testing.T) { + var got string + mt.Until(t, func() (err error) { got, err = listed(c.module); return }) + if got != c.want { + t.Fatalf("%s lists %q, want %q", c.module, got, c.want) + } + }) + } + t.Run("a ledger bound later brings them", func(t *testing.T) { + plain.Bindings = owner.Bindings + mesh.Issue(t, plain) + mt.Until(t, func() error { + got, err := listed("plain") + if err == nil && !strings.Contains(got, "ledger_rebuild:acts") { + err = errors.New("listed " + got) + } + return err + }) + }) +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index de7f3a4..8df61b5 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -52,6 +52,8 @@ type ListedTool struct { Description string `json:"description"` Input json.RawMessage `json:"input"` Subjects []string `json:"subjects,omitempty"` + // Effect is the tool's mark where the runtime knows it (novox/hq ADR 0387 §1): its own tools, the ledger repairs. + Effect string `json:"effect,omitempty"` } // NodeRuntimeModule is the module that is a machine's runtime, serving every module on it. @@ -133,6 +135,60 @@ type registration struct { module string // the module's own name, or a seat's owner string // the module whose bundle made it tools []launch.Tool + // repairs says it is the runtime's own ledger repairs for the module, not a bundle's (ledgers.go). + repairs bool +} + +// withRepairs is regs with the module's ledger repairs as its binding now says: added when it binds a ledger, removed +// when it binds none. +func withRepairs(regs []registration, repairs *registration) []registration { + out := make([]registration, 0, len(regs)+1) + for _, r := range regs { + if !r.repairs { + out = append(out, r) + } + } + if repairs != nil { + out = append(out, *repairs) + } + return out +} + +// repairsOf is a module's ledger repairs as a registration, nil when its membership binds no ledger. +func (r *runner) repairsOf(module string) *registration { + if !hasLedger(r.conn, module) { + return nil + } + return ®istration{module: module, owner: module, tools: r.events.repairTools(module), repairs: true} +} + +// refreshRepairs serves each module's ledger repairs as its membership binds a ledger now, re-serving its tools where +// that changed (a binding is written when the bus module created the ledger, design 25 § 12.2). +func (r *runner) refreshRepairs() { + r.mu.Lock() + defer r.mu.Unlock() + changed := false + for _, m := range r.modules { + l := r.of[m] + served := false + for _, reg := range l.own { + served = served || reg.repairs + } + repairs := r.repairsOf(m) + if served == (repairs != nil) { + continue + } + for i := len(l.serving) - 1; i >= 0; i-- { + l.serving[i]() + } + l.serving, l.names = nil, nil + l.own, l.regs = withRepairs(l.own, repairs), withRepairs(l.regs, repairs) + r.serve(l) + changed = true + } + if changed { + r.bindVerbs() + } } // loadedModule is one served module as its bundles were last loaded: what they registered, what of it @@ -230,6 +286,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log line += fmt.Sprintf("; not serving %s, whose bundle(s) failed to load", strings.Join(failedNames, ", ")) } logf("%s", line) + conn.OnMembership(func(bus.Membership) { go func() { r.refreshRepairs(); conn.Flush() }() }) stopSeats, again := serveSeats(conn, r.modules, r.registrations, logf) r.seatsAgain = again r.stops = append(r.stops, stopSeats) @@ -341,6 +398,9 @@ func (r *runner) load(s Served) *loadedModule { } } } + if repairs := r.repairsOf(s.Module); repairs != nil { + l.regs, l.own = append(l.regs, *repairs), append(l.own, *repairs) + } return l } @@ -411,7 +471,7 @@ func (r *runner) bindVerbs() { answer := ToolsAnswer{Module: module, Tools: []ListedTool{}, Failed: why} for _, t := range tools { answer.Tools = append(answer.Tools, ListedTool{Name: t.Name, Description: t.Description, - Input: t.Input, Subjects: subjectsOf(r.conn, module, t.Name)}) + Input: t.Input, Subjects: subjectsOf(r.conn, module, t.Name), Effect: t.Effect}) } return answer, nil }) @@ -682,6 +742,8 @@ type consumers struct { seatBound map[string]func() // provisioned is every watch a module's code holds of its values at runtime (provision.go). provisioned provisionWatches + // ledgers is how a module's ledger is opened (ledgers.go). + ledgers ledgerSource } type moduleEvents struct {