Ask a backup holder once its module's have is running, not once a send was reported (release 6, issue 1945 B8)
This commit is contained in:
@@ -80,17 +80,18 @@ func TestNatsUnassigningIrreplaceableDataRetiresItAndAnEmptyReplacementIsUrgent(
|
||||
if _, err := assign(ctx, open, "laptop", "house"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Sent, and applied: a holder is asked only once it has been (novox/hq issue 275).
|
||||
for _, node := range []string{"laptop", "anchor"} {
|
||||
pushedAndApplied(t, open, node)
|
||||
}
|
||||
|
||||
conn := onATestBus(t)
|
||||
js, err := broker.Dial(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(js.Close)
|
||||
// Applied, and running: a holder is asked only once its module's have says it runs (novox/hq issue 275, ADR 0539 §9).
|
||||
servingBus.Store(js)
|
||||
t.Cleanup(func() { servingBus.Store(nil) })
|
||||
for _, run := range [][2]string{{"laptop", "keeper"}, {"laptop", "house"}, {"anchor", "keeper"}} {
|
||||
putRunningHave(t, js, run[0], run[1])
|
||||
}
|
||||
laptop, anchor := &fakeHolder{}, &fakeHolder{}
|
||||
laptop.serve(t, conn, "laptop")
|
||||
anchor.serve(t, conn, "anchor")
|
||||
|
||||
@@ -133,3 +133,41 @@ func TestWithTheBusOutOfReachNoNodeIsSaidCurrentOrNot(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **A holder is asked once its module's have says it runs** (novox/hq ADR 0539 §9, issue 275): assigned and not applied yet, it
|
||||
// is not there to answer; a machine that has said nothing holds nothing the mesh put there.
|
||||
func TestAHolderIsAskedOnlyOnceItsModuleRuns(t *testing.T) {
|
||||
have := func(phase sdkdelivery.Phase) sdkdelivery.Have {
|
||||
return sdkdelivery.Have{Phase: phase, Health: sdkdelivery.HealthOK}
|
||||
}
|
||||
key := func(node, module string) string {
|
||||
k, err := sdkdelivery.HaveKey(node, module)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return k
|
||||
}
|
||||
nodeKey, err := sdkdelivery.HaveNodeKey("anchor")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := runningModules(map[string]sdkdelivery.Have{
|
||||
key("anchor", "keeper"): have(sdkdelivery.PhaseRunning),
|
||||
key("anchor", "house"): have(sdkdelivery.PhaseApplying),
|
||||
key("laptop", "house"): have(sdkdelivery.PhaseFailed),
|
||||
nodeKey: have(sdkdelivery.PhaseRunning),
|
||||
})
|
||||
now := time.Now()
|
||||
if !got["anchor"].settled("keeper", now) {
|
||||
t.Error("a running module's holder is not asked")
|
||||
}
|
||||
if got["anchor"].settled("house", now) {
|
||||
t.Error("a module still applying is asked")
|
||||
}
|
||||
if got["laptop"].settled("house", now) || got["ace"].settled("keeper", now) {
|
||||
t.Error("a failed module, or a machine that has said nothing, is asked")
|
||||
}
|
||||
if got["anchor"].settled("node", now) {
|
||||
t.Error("the node's own have is taken for a module")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,9 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -78,3 +81,29 @@ func (k planKeysOf) Build(_ context.Context, module, fp string) (builderkeys.Key
|
||||
}
|
||||
return builderkeys.Key{Module: module, Fingerprint: fp}, nil
|
||||
}
|
||||
|
||||
// putRunningHave writes a node-engine's have of a module: running and healthy, at a fingerprint and composition of its own.
|
||||
func putRunningHave(t *testing.T, js *broker.JetStream, node, module string) {
|
||||
t.Helper()
|
||||
ctx := t.Context()
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kv, err := api.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, err := sdkdelivery.HaveKey(node, module)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := sdkdelivery.EncodeHave(sdkdelivery.Have{FP: sdkdelivery.SourcePrefix2 + strings.Repeat("a", 64),
|
||||
Composed: "sha256:" + strings.Repeat("b", 64), Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -68,17 +68,3 @@ func pushedBy(t *testing.T, open *stores, node string) string {
|
||||
}
|
||||
return digestOf(body)
|
||||
}
|
||||
|
||||
// pushedAndApplied is pushedBy, and the machine reporting it applied that declaration.
|
||||
func pushedAndApplied(t *testing.T, open *stores, node string) {
|
||||
t.Helper()
|
||||
digest := pushedBy(t, open, node)
|
||||
record, err := open.inventory.NodeByName(t.Context(), node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := open.inventory.RecordDoing(t.Context(), record.ID, inventory.Doing{
|
||||
Outcome: inventory.OutcomeApplied, Declared: digest}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"net"
|
||||
"regexp"
|
||||
"slices"
|
||||
@@ -540,48 +541,53 @@ func holdingNeedsAnAnswer(s catalogue.Seat) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// reportGrace is how long a machine is given, after it was sent a declaration, to apply it and report
|
||||
// before what the declaration carries is expected to answer (novox/hq issue 275).
|
||||
var reportGrace = 10 * time.Minute
|
||||
|
||||
// lastSend is what a machine was last sent, as far as the self-check needs it: which modules the send
|
||||
// carried, when, and whether the machine has reported acting on it.
|
||||
// lastSend is what a machine runs, as far as the self-check needs it: the modules whose have on it is running (novox/hq ADR
|
||||
// 0539 §9). It keeps its old name, which said what was sent.
|
||||
type lastSend struct {
|
||||
// carried is the modules its last send carried; nil when that was not kept (a send from before
|
||||
// it was, or one by hand), and then every module is taken as carried — as before this existed.
|
||||
carried map[string]string
|
||||
// sent is when its current declaration went to it; nil if nothing ever did.
|
||||
sent *time.Time
|
||||
// current says its last report names the declaration last sent.
|
||||
current bool
|
||||
// running are the modules the node-engine says it runs there.
|
||||
running map[string]bool
|
||||
}
|
||||
|
||||
// settled says whether a module's holder on this machine can be asked to answer now: the machine was
|
||||
// sent a declaration carrying it, and has reported acting on that declaration or had reportGrace to.
|
||||
// A machine never sent anything holds nothing the mesh put there.
|
||||
func (d lastSend) settled(module string, now time.Time) bool {
|
||||
if d.sent == nil {
|
||||
return false
|
||||
}
|
||||
if d.carried != nil {
|
||||
if _, in := d.carried[module]; !in {
|
||||
return false
|
||||
// settled says whether a module's holder on this machine can be asked to answer now: the node-engine's have of the module is
|
||||
// running. A module assigned and not applied yet is not there to answer, and a machine that has said nothing holds nothing the
|
||||
// mesh put there. now is unused: the have says it, with no grace to wait out.
|
||||
func (d lastSend) settled(module string, _ time.Time) bool { return d.running[module] }
|
||||
|
||||
// runningModules reads from the haves which modules run on which node. Pure.
|
||||
func runningModules(haves map[string]sdkdelivery.Have) map[string]lastSend {
|
||||
out := map[string]lastSend{}
|
||||
for key, h := range haves {
|
||||
node, module, isNode, err := sdkdelivery.ParseHaveKey(key)
|
||||
if err != nil || isNode || h.Phase != sdkdelivery.PhaseRunning {
|
||||
continue
|
||||
}
|
||||
d := out[node]
|
||||
if d.running == nil {
|
||||
d.running = map[string]bool{}
|
||||
}
|
||||
d.running[module] = true
|
||||
out[node] = d
|
||||
}
|
||||
return d.current || now.Sub(*d.sent) > reportGrace
|
||||
return out
|
||||
}
|
||||
|
||||
// readDeliveries is every machine's last send, by name, from the records a send and a report keep.
|
||||
// readDeliveries is every machine's running modules, by name, from the node-engines' haves. A process with no bus reads none:
|
||||
// nothing is then taken as running, never everything.
|
||||
func readDeliveries(ctx context.Context, inv *inventory.Inventory) (map[string]lastSend, error) {
|
||||
reports, err := lastReports(ctx, inv)
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
return map[string]lastSend{}, nil // empty-on-error: no bus to read the haves from, so no module is taken as running
|
||||
}
|
||||
defer bus.Close()
|
||||
js, err := jetstream.New(bus.Conn())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := map[string]lastSend{}
|
||||
for _, r := range reports {
|
||||
out[r.Node] = lastSend{sent: r.Sent, current: r.Current}
|
||||
_, haves, err := readWantsAndHaves(ctx, js)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
return runningModules(haves), nil
|
||||
}
|
||||
|
||||
// heardMachines is every machine within its heartbeat's bound, as the watchdogs last saw.
|
||||
|
||||
Reference in New Issue
Block a user