diff --git a/cmd/mesh-host/applywhole_test.go b/cmd/mesh-host/applywhole_test.go new file mode 100644 index 0000000..f332756 --- /dev/null +++ b/cmd/mesh-host/applywhole_test.go @@ -0,0 +1,41 @@ +package main + +import ( + "context" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/novox/mesh-host/internal/apply" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/store" +) + +// applyAndKeep applies a whole declaration the way the carried bundle is applied, and says what came of it as the facts +// the node says: what the tests of the apply's steps read. The node-engine applies no whole declaration on its own now +// (novox/hq ADR 0539 §9); this is the test's way to put a machine in a state. +func applyAndKeep(ctx context.Context, opts options, raw []byte, signed *store.Declared, sched *apply.Scheduler, + say link.Announce) link.Report { + declared, err := declaration.Parse(raw) + if err != nil { + return link.Report{Refused: err.Error()} + } + got := applyModuleHeld(ctx, opts, declared, nil, sched, say) + if got.Refused != "" { + return link.Report{Refused: got.Refused} + } + held := map[string]bool{} + for _, h := range got.Updated.Held { + held[h.ID] = true + } + report := factsOf(ctx, opts, declared, got.Updated, got.Outcome, held) + if signed != nil { + if err := store.SaveDeclared(store.DeclaredPath(opts.state), *signed); err != nil { + report.Failed = map[string]string{"keeping the declaration": err.Error()} + } + } + if got.Err != nil { + report.Failed = map[string]string{"apply": got.Err.Error()} + } + report.Failed = withUnanswered(report.Failed, got.Probe) + report.Unapplied = unapplied(declared, got.Outcome, got.Err) + return report +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 94ed226..3aef809 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -30,7 +30,9 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/declaration" + "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-host/internal/accounts" + "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/bundle" "github.com/novox/mesh-host/internal/firewall" @@ -1158,8 +1160,12 @@ func runLink(ctx context.Context, opts options) error { Say: say, Timeout: opts.timeout, } + current := &agentNow{} + factsOfNow := func(ctx context.Context, ag *agent.Agent) { + sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag) + } queue.OnLink = func(ctx context.Context) { - sayFacts(ctx, opts, mine.Membership.Signer, queue, say) + factsOfNow(ctx, current.get()) afterApply() } // And the move of a value sealed to this machine into the vault, asked by the controller (design 57's V3, @@ -1186,6 +1192,14 @@ func runLink(ctx context.Context, opts options) error { go judging.Probe(aside) go judgeWhatRuns(aside, judging, queue, say) } + // The node follows its wants: one module at a time from the composition its want names (novox/hq ADR 0539), and the + // busy containers are asked for again soon (ADR 0360 §11). + go nodeAgent(aside, opts, mine.Membership.Signer, queue, sched, say, afterApply, current, factsOfNow) + go busyPasses(aside, func() { + if ag := current.get(); ag != nil { + ag.Retry(busyModules()...) + } + }, BusyAskEvery) // And the core builds this host placed are judged, whichever host placed them (to-be 45 §8). go watchWhatThisHostPlaced(aside, mine.Node, queue, say) // And mesh-cli is answered on this machine: the account from the kernel, the line asked of the controller @@ -1563,6 +1577,23 @@ func orNothing(s, unless string) string { // busyWaiting is whether the last apply left a container running because it said busy (novox/hq ADR 0335). var busyWaiting atomic.Bool +// busyBy is, per module, whether its last apply left a container running because it said busy; busyWaiting is any of them. +var ( + busyMu sync.Mutex + busyBy = map[string]bool{} +) + +func noteBusy(module string, waiting bool) { + busyMu.Lock() + defer busyMu.Unlock() + busyBy[module] = waiting + any := false + for _, w := range busyBy { + any = any || w + } + busyWaiting.Store(any) +} + // BusyAskEvery is how often a container waiting because it said busy is asked again (novox/hq ADR 0335 §2): // each ask is a pass of its own, so a build that ends is followed by its replacement within this, not // within the reconcile's five minutes — and the bound of 15 minutes is met within this of its end. @@ -1611,100 +1642,86 @@ func announceOr(say link.Announce) func(string) { // run by hand beside this service — the lock beside the state does the same (store.Lock). var applying sync.Mutex -// applyAndKeep applies a declaration and, when it came from the mesh, keeps it so this node can -// go on obeying it while disconnected. One at a time, whoever asks. Given unsigned, the bytes are -// what this node kept, already verified against the mesh's key on load. -func applyAndKeep(ctx context.Context, opts options, raw []byte, signed *store.Declared, - sched *apply.Scheduler, say link.Announce) link.Report { - applying.Lock() - defer applying.Unlock() - return applyAndKeepHeld(ctx, opts, raw, signed, sched, say) +// appliedModule is how the apply of one module's declaration came out. +type appliedModule struct { + // Refused is why nothing was applied, when nothing was. + Refused string + Outcome apply.Report + Updated store.State + // Err is the apply's own error, when it failed part way. + Err error + Probe *unanswered } -// applyAndKeepHeld is applyAndKeep for a caller already holding applying. -func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *store.Declared, - sched *apply.Scheduler, say link.Announce) link.Report { +// applyModuleHeld applies one module's declaration (novox/hq ADR 0539 §9): the resources of its composition, with the +// compositions the node's other modules run held out of any removal. A nil scope is a whole declaration, as the carried +// bundle is. One at a time, and the state is read and saved under the lock of the node's state. +func applyModuleHeld(ctx context.Context, opts options, declared *declaration.Declaration, scope *apply.Scope, + sched *apply.Scheduler, say link.Announce) appliedModule { + applying.Lock() + defer applying.Unlock() unlock, err := store.Lock(opts.state, nil) if err != nil { - return link.Report{Refused: err.Error()} + return appliedModule{Refused: err.Error()} } defer unlock() - declared, err := declaration.Parse(raw) - if err != nil { - return link.Report{Refused: err.Error()} - } known, err := store.Load(opts.state) if err != nil { - return link.Report{Refused: err.Error()} + return appliedModule{Refused: err.Error()} } - // A declaration from the mesh is the controller's word on the node's mode — the flip arrives - // as exactly that, the first converged declaration after adopted ones — and becomes the - // record. So does what this node kept: it is signed by the mesh and verified on load, where - // the state's note of the mode is this host's own. Where they disagree the note is wrong — a - // genesis re-run over a node the mesh converged since, a state saved when the kept declaration - // could not be — and it is repaired and said, not obeyed: refusing would hold this node off what - // the mesh said until a delivery that comes only when something changes (novox/hq issue 104). - if signed == nil && known.Mode != "" && known.Mode != apply.ModeOf(declared) { - if say != nil { - say(fmt.Sprintf("this node's record said %s; what the mesh last said, signed, says %s — the record is repaired", - known.Mode, apply.ModeOf(declared))) - } + node := scope == nil || scope.Module == store.NodeModule + // The node's own composition is the controller's word on the node's mode — the flip arrives as exactly that, the + // first converged composition after adopted ones — and becomes the record (novox/hq issue 104). + if node && known.Mode != "" && known.Mode != apply.ModeOf(declared) { + announceOr(say)(fmt.Sprintf("this node's record said %s; what the mesh last said, signed, says %s — the record is repaired", + known.Mode, apply.ModeOf(declared))) known.Mode = apply.ModeOf(declared) } - // **A file given by digest is fetched and verified before anything of the declaration is applied** (novox/hq ADR - // 0486 §1, issue 1695): one object missing or differing refuses the declaration whole. The signed bytes are what is - // kept; only what is applied has the content in place of the reference. + // **A file given by digest is fetched and verified before anything of the composition is applied** (novox/hq ADR + // 0486 §1, issue 1695): one object missing or differing refuses the module whole. if err := apply.ResolveFileReferences(ctx, declared, link.FilesFrom(ctx)); err != nil { - return link.Report{Refused: "a file of this declaration could not be had by its digest: " + err.Error()} + return appliedModule{Refused: "a file of this composition could not be had by its digest: " + err.Error()} } built, err := system.For(builtFor) if err != nil { - return link.Report{Refused: err.Error()} + return appliedModule{Refused: err.Error()} } if err := system.Check(built, declared); err != nil { - return link.Report{Refused: err.Error()} + return appliedModule{Refused: err.Error()} } - // A package manager that does not answer fails only what needs it, and the report says so (novox/hq - // issue 233, probe.go). + // A package manager that does not answer fails only what needs it, and the report says so (novox/hq issue 233). probe, err := probePackageManager(ctx, built) if err != nil { - return link.Report{Refused: err.Error()} + return appliedModule{Refused: err.Error()} } if probe != nil { announceOr(say)(probe.said()) } - // Declared, not carried. A declaration from the mesh removes only what the mesh previously - // declared — never what this machine raised for itself from its bundle (04-ISSUES/010). - // **What the apply says goes to the console, which is the journal when this runs as a service.** - // - // It was nil, and nil is silence. The one-shot path has always passed a real one, so every detail - // the apply produces — a file held, a container replaced, the found firewall retired — was visible - // when a person ran it by hand and discarded in the way the host actually runs. Measured: after a - // machine was converged and its found firewall was not retired, what the host decided was - // unrecoverable, because it had said it to nobody (novox/hq 04-ISSUES/143). - // - // `say` already reaches stdout, and the launcher's unit sends that to the journal, so this needs - // no new mechanism — only for the argument to be passed. - // Once more when the package manager did not answer and this apply wrote something (issue 233, probe.go). + // **What the apply says goes to the console, which is the journal when this runs as a service** (novox/hq + // 04-ISSUES/143). Once more when the package manager did not answer and this apply wrote something (issue 233). outcome, updated, applyErr, probe, firstPass := applyMindingProbe(ctx, built, probe, known, func(known store.State) (apply.Report, store.State, error) { - return apply.ApplyMindingWindows(ctx, built, declared, known, store.OriginDeclared, + if scope == nil { + return apply.ApplyMindingWindows(ctx, built, declared, known, store.OriginDeclared, + apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), + apply.WindowsIn(filepath.Dir(opts.state))) + } + return apply.ApplyModule(ctx, built, declared, *scope, known, store.OriginDeclared, apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state))) }, announceOr(say)) - // The mode the mesh said, recorded whichever way the apply went: the declaration is kept - // either way, and the node is held to it from the next reconcile (novox/hq ADR 0100). - updated.Mode = apply.ModeOf(declared) + if node { + updated.Mode = apply.ModeOf(declared) + } - // Root acted on what an agent account's judgement reads — an account, a package, an action, sudo's - // rules: the search for setuid programs runs again (novox/hq issue 361). A failed apply counts what it - // did before it stopped. + // Root acted on what an agent account's judgement reads — an account, a package, an action, sudo's rules: the search + // for setuid programs runs again (novox/hq issue 361). A failed apply counts what it did before it stopped. moved := append(append([]apply.Outcome(nil), firstPass...), outcome.Outcomes...) var partial *apply.Error if errors.As(applyErr, &partial) { @@ -1714,55 +1731,34 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto setuidSearch.SearchAgain() } - // Saved whichever way it went. Recording only on success would lose the footprint of a - // failed apply, and that footprint is on the machine either way. + // Saved whichever way it went: the footprint of a failed apply is on the machine either way. if saveErr := store.Save(opts.state, updated); saveErr != nil { - return link.Report{Refused: "applied, and the node's state could not be saved: " + - saveErr.Error()} + return appliedModule{Refused: "applied, and the node's state could not be saved: " + saveErr.Error()} } - // Re-establish the scheduled steps from the declaration just applied (novox/hq ADR 0053). Done - // from the parsed declaration, which is the source of truth (ADR 0018): a schedule newly declared - // is armed, one whose image, environment or cadence changed is re-armed, and one no longer - // declared is forgotten — and after a host restart the first apply rebuilds them all. A nil - // scheduler is the one-shot CLI path, which exits rather than staying up to fire anything. + // Re-establish the scheduled steps of this declaration (novox/hq ADR 0053): a schedule newly declared is armed, one + // whose image, environment or cadence changed is re-armed, one no longer declared is forgotten, and the other + // modules' are left as they are. A nil scheduler is a test. held := map[string]bool{} for _, h := range updated.Held { held[h.ID] = true } if sched != nil { - sched.Sync(declared, held) - } - - report := factsOf(ctx, opts, declared, updated, outcome, held) - // Kept whichever way it went, so a node that is disconnected next minute still knows what it - // was told. Saved after applying rather than before: what is kept is what this node acted on. - var before []string - if signed != nil { - // The modules the declaration kept until now held, so a have key written for one the new declaration no longer - // holds is written removed even by a node-engine that started since (issue 1369). - if kept, keptErr := store.ReadDeclared(store.DeclaredPath(opts.state)); keptErr == nil { - if old, parseErr := declaration.ParseTrusted(kept.Declaration); parseErr == nil { - for _, f := range moduleFacts(old, apply.Report{}, nil) { - before = append(before, f.Module) - } - } - } - if err := store.SaveDeclared(store.DeclaredPath(opts.state), *signed); err != nil { - report.Failed = map[string]string{"keeping the declaration": err.Error()} + if scope == nil { + sched.Sync(declared, held) + } else { + sched.SyncModule(declared, held, func(id string) bool { + m, ok := declaration.ModuleGuess(id) + return ok && m == scope.Module || (!ok && scope.Module == store.NodeModule) + }) } } - if applyErr != nil { - report.Failed = map[string]string{"apply": applyErr.Error()} + name := "" + if scope != nil { + name = scope.Module } - report.Failed = withUnanswered(report.Failed, probe) - // And how each module of the declaration came out, for the have keys written beside this report (novox/hq ADR 0429 - // §4, issue 1132): not on the wire. - report.Modules = moduleFacts(declared, outcome, applyErr) - report.Before = before - // Which resources did not apply, for whether a reconcile's account is news (issue 233): not said on the wire. - report.Unapplied = unapplied(declared, outcome, applyErr) - return report + noteBusy(name, outcome.Waiting()) + return appliedModule{Outcome: outcome, Updated: updated, Err: applyErr, Probe: probe} } // factsOf is what this node says of itself: its ports, its profile, how what runs here is, which links face outside, @@ -1896,16 +1892,22 @@ func factsOf(ctx context.Context, opts options, declared *declaration.Declaratio return report } -// sayFacts says this node's facts on the link open now (novox/hq ADR 0539 §9): what the node is made of, how what runs here -// is, and the haves of the modules the declaration it last applied holds. It applies nothing. -func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, session *link.Session, say link.Announce) { +// sayFacts says this node's facts on the link open now (novox/hq ADR 0539 §9): what the node is made of, how what runs +// here is, and its filters and ports. What runs is the compositions the node agent holds running (ag), or, before it +// has run, the declaration the node kept before the switch. It applies nothing; the haves are the agent's. +func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, session *link.Session, say link.Announce, + ag *agent.Agent) { known, err := store.Load(opts.state) if err != nil { say("cannot say this node's facts: " + err.Error()) return } declared := &declaration.Declaration{} - if raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer); err == nil { + if running := runningOf(ag); len(running) > 0 { + if d, err := declarationOf(running...); err == nil { + declared = d + } + } else if raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer); err == nil { if d, err := declaration.Parse(raw); err == nil { declared = d } @@ -1918,6 +1920,13 @@ func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, sessi session.SayFacts(ctx, report) } +func runningOf(ag *agent.Agent) []delivery.Composed { + if ag == nil { + return nil + } + return ag.Running() +} + // waitForEnrolment returns this node's identity, waiting for one if it has none. // // It applies the carried bundle first, if there is one, because that is what a first node does diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go new file mode 100644 index 0000000..611c74b --- /dev/null +++ b/cmd/mesh-host/nodeagent.go @@ -0,0 +1,353 @@ +package main + +import ( + "context" + "crypto/ed25519" + "encoding/json" + "fmt" + "sort" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/apply" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/store" +) + +// The node agent's wiring (novox/hq ADR 0429 §4, ADR 0539 §9): the buckets it watches, the haves it writes, the backup +// it asks and the apply of one module's composition. + +// bucketWatch is a bucket of the bus as the agent watches it, through the session's link. +type bucketWatch struct { + session *link.Session + bucket string +} + +func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Update, error) { + in, err := w.session.KVWatch(ctx, w.bucket, pattern) + if err != nil { + return nil, err + } + out := make(chan agent.Update, 64) + go func() { + defer close(out) + for u := range in { + select { + case out <- agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted, Caught: u.Caught}: + case <-ctx.Done(): + return + } + } + }() + return out, nil +} + +// sessionHaves writes the haves the agent says, through the session. +type sessionHaves struct{ session *link.Session } + +func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) { + got, ok := h.session.ReadHave(ctx, module) + if !ok { + return agent.Have{}, false + } + return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail}, true +} + +func (h sessionHaves) Write(ctx context.Context, x agent.Have) { + if x.Module == agent.NodeModule { + // The node's own key holds its facts and the digest of the composition it runs. + if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) { + h.session.NoteNode(ctx, x.Composed) + } + return + } + phase := x.Phase + if phase == string(delivery.PhaseRunning) { + phase = "" + } + h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase, + Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail)}) +} + +func (h sessionHaves) Removed(ctx context.Context, module string) { + h.session.NoteRemoved(ctx, module, "") +} + +// shorten keeps a have's detail within what a have value holds. +func shorten(s string) string { + const most = 300 + if len(s) <= most { + return s + } + return s[:most-1] + "…" +} + +// backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14), and backupPollEvery how +// soon the holder is asked again how it went while it works. +var ( + backupWithin = 10 * time.Minute + backupPollEvery = 5 * time.Second +) + +// backupNode asks the holder of node-backup on this node to back a module up now, and waits for it to say how it went: +// the holder answers when it has started, and its `backed-up` says the module's last night. +func backupNode(session *link.Session) func(ctx context.Context, module string) agent.Backup { + return func(ctx context.Context, module string) agent.Backup { + asked := time.Now() + if _, err := session.AskSeat(ctx, "node-backup", "now", map[string]any{"module": module}, 30*time.Second); err != nil { + return agent.Backup{Words: err.Error()} + } + deadline := asked.Add(backupWithin) + for { + select { + case <-ctx.Done(): + return agent.Backup{Words: "the node agent was stopped while the backup ran"} + case <-time.After(backupPollEvery): + } + raw, err := session.AskSeat(ctx, "node-backup", "backed-up", map[string]any{"module": module}, 10*time.Second) + if err == nil { + var read []struct { + Module string `json:"module"` + LastNight *struct { + OK bool `json:"ok"` + At time.Time `json:"at"` + Error string `json:"error"` + } `json:"lastNight"` + } + if json.Unmarshal(raw, &read) == nil { + for _, m := range read { + if m.Module == module && m.LastNight != nil && m.LastNight.At.After(asked) { + if m.LastNight.OK { + return agent.Backup{Done: true} + } + return agent.Backup{Words: "the backup failed: " + m.LastNight.Error} + } + } + } + } + if time.Now().After(deadline) { + return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", backupWithin)} + } + } + } +} + +// declarationOf is the declaration the compositions make together: their resources, and the node's own parts when the +// node's composition is among them. It is what the node judges and says its health by, and what one module's apply is +// made of (with a single composition). +func declarationOf(cs ...delivery.Composed) (*declaration.Declaration, error) { + d := &declaration.Declaration{Version: declaration.Version} + for _, c := range cs { + for _, raw := range c.Resources { + r, err := declaration.ParseResource(raw) + if err != nil { + return nil, fmt.Errorf("a resource of %q: %w", c.Module, err) + } + d.Resources = append(d.Resources, r) + } + d.Filled = append(d.Filled, c.Filled...) + if c.Module == "" { + d.Adoption = c.Own.Adoption + d.Identities = append(d.Identities, c.Own.Identities...) + d.Groups = append(d.Groups, c.Own.Groups...) + d.Libraries = append(d.Libraries, c.Own.Libraries...) + d.Provisioner = c.Own.Provisioner + } + } + return d, nil +} + +func scopeName(c delivery.Composed) string { + if c.Module == "" { + return agent.NodeModule + } + return c.Module +} + +func resourcesOf(cs []delivery.Composed) []declaration.Resource { + var out []declaration.Resource + for _, c := range cs { + for _, raw := range c.Resources { + if r, err := declaration.ParseResource(raw); err == nil { + out = append(out, r) + } + } + } + return out +} + +// keptByModule is the resources of the declaration the node kept before the switch, by the module they say; the node's +// own are those that say none (ADR 0539 §9). +func keptByModule(opts options, signer ed25519.PublicKey) func(module string) ([]json.RawMessage, bool) { + var once sync.Once + var by map[string][]json.RawMessage + return func(module string) ([]json.RawMessage, bool) { + once.Do(func() { + raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer) + if err != nil { + return + } + var kept struct { + Resources []json.RawMessage `json:"resources"` + } + if json.Unmarshal(raw, &kept) != nil { + return + } + by = map[string][]json.RawMessage{} + for _, r := range kept.Resources { + var head struct { + Module string `json:"module"` + } + _ = json.Unmarshal(r, &head) + name := head.Module + if name == "" { + name = agent.NodeModule + } + by[name] = append(by[name], r) + } + }) + r, ok := by[module] + return r, ok + } +} + +// nodeAgent runs the node agent under the session's links for as long as ctx lasts: it waits for a link, follows the +// node's wants and compositions on it, and begins again with the next link. A watch it cannot make (a bucket this node's +// account may not read) is said, and tried again with a growing wait, as the link is. +func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, session *link.Session, sched *apply.Scheduler, + say link.Announce, afterApply func(), current *agentNow, sayFacts func(ctx context.Context, ag *agent.Agent)) { + wait := 2 * time.Second + for ctx.Err() == nil { + linked, cancel, err := session.WhenLinked(ctx) + if err != nil { + return + } + began := time.Now() + node := session.Membership.Node + composedBucket, err := delivery.ComposedBucketFor(node) + if err != nil { + cancel() + say("the node agent cannot name this node's composition bucket: " + err.Error()) + return + } + var ag *agent.Agent + env := agent.Env{ + Node: node, + Wants: bucketWatch{session, delivery.AssignerWantsBucket}, + Composed: bucketWatch{session, composedBucket}, + Haves: sessionHaves{session}, + Kept: keptByModule(opts, signer), + Backup: backupNode(session), + Say: say, + Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { + res := applyComposition(session.WithLink(ctx), opts, c, others, sched, say) + afterApply() + sayFacts(ctx, ag) + return res + }, + Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result { + res := removeComposition(session.WithLink(ctx), opts, module, others, sched, say) + afterApply() + sayFacts(ctx, ag) + return res + }, + } + ag = agent.New(env) + current.set(ag) + err = ag.Run(linked) + cancel() + current.set(nil) + if ctx.Err() != nil { + return + } + if time.Since(began) > 30*time.Second { + wait = 2 * time.Second + } + if err != nil { + say(fmt.Sprintf("the node agent stopped: %v — beginning again in %s", err, wait)) + } + select { + case <-ctx.Done(): + return + case <-time.After(wait): + } + if wait *= 2; wait > 2*time.Minute { + wait = 2 * time.Minute + } + } +} + +// agentNow is the agent running now, which the node's facts are built from. +type agentNow struct { + mu sync.Mutex + ag *agent.Agent +} + +func (c *agentNow) set(a *agent.Agent) { + c.mu.Lock() + c.ag = a + c.mu.Unlock() +} + +func (c *agentNow) get() *agent.Agent { + c.mu.Lock() + defer c.mu.Unlock() + return c.ag +} + +// applyComposition applies one composition: a module's, or the node's own. +func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed, + sched *apply.Scheduler, say link.Announce) agent.Result { + name := scopeName(c) + d, err := declarationOf(c) + if err != nil { + return agent.Result{Failed: true, Detail: shorten(err.Error())} + } + scope := apply.Scope{Module: name, Others: resourcesOf(others)} + got := applyModuleHeld(ctx, opts, d, &scope, sched, say) + if got.Refused != "" { + return agent.Result{Failed: true, Detail: shorten(got.Refused)} + } + results := apply.ModuleResults(d, got.Outcome, got.Err) + if r, ok := results[name]; ok && r.Failed { + return agent.Result{Failed: true, Detail: shorten(r.Detail)} + } + if got.Err != nil { + return agent.Result{Failed: true, Detail: shorten(got.Err.Error())} + } + if got.Probe != nil { + return agent.Result{Failed: true, Detail: shorten(got.Probe.partial().Error())} + } + return agent.Result{} +} + +// removeComposition removes one module's own resources and nothing of another's. +func removeComposition(ctx context.Context, opts options, module string, others []delivery.Composed, + sched *apply.Scheduler, say link.Announce) agent.Result { + scope := apply.Scope{Module: module, Others: resourcesOf(others)} + got := applyModuleHeld(ctx, opts, &declaration.Declaration{Version: declaration.Version}, &scope, sched, say) + switch { + case got.Refused != "": + return agent.Result{Failed: true, Detail: shorten(got.Refused)} + case got.Err != nil: + return agent.Result{Failed: true, Detail: shorten(got.Err.Error())} + } + return agent.Result{} +} + +// busyModules are the modules whose last apply left a container running because it said busy. +func busyModules() []string { + busyMu.Lock() + defer busyMu.Unlock() + var out []string + for m, w := range busyBy { + if w && m != "" { + out = append(out, m) + } + } + sort.Strings(out) + return out +} diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 471b0d7..12e8375 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -107,6 +107,7 @@ type Agent struct { // prior is a module the node ran before this agent started, so it has something to back up. prior map[string]bool said map[string]string + retry chan []string } type wantNow struct { @@ -137,7 +138,8 @@ func New(env Env) *Agent { env.Say = func(string) {} } return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, - attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}} + attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, + retry: make(chan []string, 8)} } // Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition @@ -157,6 +159,16 @@ func (a *Agent) Run(ctx context.Context) error { select { case <-ctx.Done(): return nil + case modules := <-a.retry: + // Applied again though the keys did not change: a container that waited because it said busy (ADR 0360 + // §11) is replaced once it is idle. + a.mu.Lock() + for _, m := range modules { + delete(a.running, m) + delete(a.attempt, m) + dirty[m] = true + } + a.mu.Unlock() case u, ok := <-wants: if !ok { return fmt.Errorf("the watch of the wants ended") @@ -186,6 +198,39 @@ func (a *Agent) Run(ctx context.Context) error { } } +// Retry asks for the modules to be applied again from the keys they have, though nothing changed: what ADR 0360 §11 +// needs for a container that waits because it said busy. It never blocks; a retry asked while eight wait is dropped, and +// asked again by its caller's next look. +func (a *Agent) Retry(modules ...string) { + select { + case a.retry <- modules: + default: + } +} + +// Running is the compositions the node runs now: each module's that the agent applied or found already running, and +// the node's own. What the node judges its health by and says of itself is built from them. +func (a *Agent) Running() []delivery.Composed { + a.mu.Lock() + defer a.mu.Unlock() + mods := make([]string, 0, len(a.running)) + for m := range a.running { + mods = append(mods, m) + } + sort.Strings(mods) + var out []delivery.Composed + for _, m := range mods { + h := a.running[m] + if h.Phase == delivery.PhaseRemoved { + continue + } + if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed { + out = append(out, c.C) + } + } + return out +} + // takeWant takes a change of a want key and returns the module it concerns, or "" for a key that is not a want. func (a *Agent) takeWant(u Update) string { a.mu.Lock() diff --git a/internal/apply/schedule.go b/internal/apply/schedule.go index 9b7146b..f40ee63 100644 --- a/internal/apply/schedule.go +++ b/internal/apply/schedule.go @@ -127,6 +127,17 @@ func (s *Scheduler) RecordWindowsIn(w *Windows) { // work on the predecessor's data, under the predecessor's service, which is the one thing an // adopted node must not do. Nil on a converged node, where nothing is held. func (s *Scheduler) Sync(d *declaration.Declaration, held map[string]bool) { + s.sync(d, held, nil) +} + +// SyncModule is Sync for one module's declaration (novox/hq ADR 0539 §9): the schedules of that module are armed, +// re-armed or forgotten, and the schedules of every other module are left as they are. owns says whether a scheduled +// step's id is the module's. +func (s *Scheduler) SyncModule(d *declaration.Declaration, held map[string]bool, owns func(id string) bool) { + s.sync(d, held, owns) +} + +func (s *Scheduler) sync(d *declaration.Declaration, held map[string]bool, owns func(id string) bool) { s.mu.Lock() defer s.mu.Unlock() @@ -177,7 +188,7 @@ func (s *Scheduler) Sync(d *declaration.Declaration, held map[string]bool) { } for id := range s.jobs { - if !seen[id] { + if !seen[id] && (owns == nil || owns(id)) { delete(s.jobs, id) } } diff --git a/internal/link/have.go b/internal/link/have.go index db350cd..caa01c1 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -420,3 +420,35 @@ func (q *Session) NoteNode(ctx context.Context, composed string) { q.mu.Unlock() q.writeHaves(ctx, bus) } + +// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot +// be read answers as none: the agent then compares with the declaration it kept, and applies only what differs. +func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + hb, ok := bus.(HaveBus) + if !ok || bus == nil { + return delivery.Have{}, false + } + key, err := delivery.HaveKey(q.Membership.Node, module) + if err != nil { + return delivery.Have{}, false + } + read, cancel := context.WithTimeout(ctx, q.timeout()) + defer cancel() + bucket, err := hb.HaveBind(read) + if err != nil { + return delivery.Have{}, false + } + value, _, err := bucket.Get(read, key) + if err != nil { + return delivery.Have{}, false + } + h, err := delivery.DecodeHave(value) + if err != nil { + return delivery.Have{}, false + } + return h, true +} diff --git a/internal/link/session.go b/internal/link/session.go index 0eb2d80..541f773 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -54,6 +54,9 @@ type Session struct { mu sync.Mutex bus Bus // the link open now; nil while there is none + // gone is closed when the link open now closes, and changed whenever a link opens or closes (WhenLinked). + gone chan struct{} + changed chan struct{} // saying is held while a facts event is made and said, so they leave in the order they are made. saying sync.Mutex @@ -72,6 +75,7 @@ func (q *Session) init() { q.Say = func(string) {} } q.epoch = time.Now().UnixMilli() + q.changed = make(chan struct{}) }) } @@ -117,6 +121,9 @@ func (q *Session) attach(ctx context.Context, bus Bus) { q.init() q.mu.Lock() q.bus = bus + q.gone = make(chan struct{}) + close(q.changed) + q.changed = make(chan struct{}) q.mu.Unlock() q.FactsAsked(ctx) } @@ -141,6 +148,9 @@ func (q *Session) detach(bus Bus) { q.mu.Lock() if q.bus == bus { q.bus = nil + close(q.gone) + close(q.changed) + q.changed = make(chan struct{}) } q.mu.Unlock() } @@ -255,3 +265,46 @@ func ReadToolAnswer(reply []byte) (health.Answer, error) { } return a, nil } + +// SeatToolSubject is where a node seat's verb is asked of one machine's holder (design 33 §4): the seat's verb subject +// with the machine as its last token. +func SeatToolSubject(seat, verb, node string) string { + return "mesh.seat." + seat + ".tool." + verb + "." + node +} + +// AskSeat asks the holder of a seat on this node a verb, over the link open now, and answers its result, or its refusal +// as an error. The one call the node agent makes of another part: the backup before a data module moves (ADR 0429 §14). +func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, within time.Duration) (json.RawMessage, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + if bus == nil { + return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked) + } + asker, ok := bus.(ToolBus) + if !ok { + return nil, fmt.Errorf("%w: the link to the bus open now cannot ask a seat", ErrNotAsked) + } + body, err := json.Marshal(args) + if err != nil { + return nil, err + } + asking, cancel := context.WithTimeout(ctx, within) + defer cancel() + reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, q.Membership.Node), body) + if err != nil { + return nil, fmt.Errorf("%s.%s could not be asked: %w", seat, verb, err) + } + var env struct { + Result json.RawMessage `json:"result"` + Error string `json:"error"` + } + if err := json.Unmarshal(reply, &env); err != nil { + return nil, fmt.Errorf("%s.%s answered with something that is not a tool's answer", seat, verb) + } + if env.Error != "" { + return nil, errors.New(env.Error) + } + return env.Result, nil +} diff --git a/internal/link/watch.go b/internal/link/watch.go new file mode 100644 index 0000000..8b1c123 --- /dev/null +++ b/internal/link/watch.go @@ -0,0 +1,129 @@ +package link + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go" +) + +// KVUpdate is one change of a key in a bucket the node-engine watches (novox/hq ADR 0539 §9). Caught marks the end of +// the keys that were there when the watch began. +type KVUpdate struct { + Key string + Value []byte + Revision uint64 + Deleted bool + Caught bool +} + +// KVBus is a link that can watch a bucket's keys: the node agent follows its wants and its compositions with it, and +// never polls. +type KVBus interface { + // KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then every change, until ctx ends + // or the link does (the channel is then closed). A bucket the node's account may not read is an error. + KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) +} + +// KVWatch is the watch on the link open now. +func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) { + js := b.JS + if b.Conn != nil { + bound, err := b.Conn.JetStream() + if err != nil { + return nil, err + } + js = bound + } + kv, err := js.KeyValue(bucket) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", bucket, err) + } + w, err := kv.Watch(pattern, nats.Context(ctx)) + if err != nil { + return nil, fmt.Errorf("watching %s in %s: %w", pattern, bucket, err) + } + out := make(chan KVUpdate, 64) + go func() { + defer close(out) + defer func() { _ = w.Stop() }() + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + var u KVUpdate + if e == nil { + u.Caught = true + } else { + u = KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(), + Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge} + } + select { + case out <- u: + case <-ctx.Done(): + return + } + } + } + }() + return out, nil +} + +func (l *natsLink) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) { + return OverNATS{Conn: l.conn, JS: l.js}.KVWatch(ctx, bucket, pattern) +} + +var ( + _ KVBus = OverNATS{} + _ KVBus = (*natsLink)(nil) +) + +// ErrNoLink is what a watch asks for with no link open. +var ErrNoLink = errors.New("no link to the bus is open") + +// KVWatch watches a bucket on the link open now. It ends when ctx does or the link closes. +func (q *Session) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + w, ok := bus.(KVBus) + if bus == nil || !ok { + return nil, ErrNoLink + } + return w.KVWatch(ctx, bucket, pattern) +} + +// WhenLinked waits for a link to be open and returns a context that ends when that link does (or ctx does): what the +// node agent runs under, so its watches end with the link and begin again with the next. +func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.CancelFunc, error) { + q.init() + for { + q.mu.Lock() + if q.bus != nil { + gone := q.gone + q.mu.Unlock() + linked, cancel := context.WithCancel(ctx) + go func() { + select { + case <-gone: + cancel() + case <-linked.Done(): + } + }() + return linked, cancel, nil + } + changed := q.changed + q.mu.Unlock() + select { + case <-ctx.Done(): + return nil, nil, ctx.Err() + case <-changed: + } + } +}