novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

indexer: a wanted that says the want removed empties that member from named

Browse Source 
A member a running put in a set leaves it also when the want of that node and module is removed (the node left or the module was
unplaced), whatever that node last said running; a running or a wanted older than the removal does not bring it back, so a
restart that reads the two streams in either order reaches the same sets.

Builds: design 47 G §675, §693
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 23:00:14 +01:00
parent 86bae053b6
commit d2441d5fcc
3 changed files with 43 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+12 -3
View File
@@ -1433,9 +1433,10 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
// TestIndexerKeepsWhatAWantAndARunningBuildName (design 47 G §675, §693, A-61): a want said before the start on
// wanted.<node>.<module> and a running build of another node are read whole at the start into the state named.
func TestIndexerKeepsWhatAWantAndARunningBuildName(t *testing.T) {
// TestIndexerKeepsWhatAWantAndARunningBuildNameAndLetsAGoneOneLeave (design 47 G §675, §693, A-61): a want said before the start on
// wanted.<node>.<module> and a running build of another node are read whole at the start into the state named; a later
// wanted that says the want removed empties its member whatever the node last said running.
func TestIndexerKeepsWhatAWantAndARunningBuildNameAndLetsAGoneOneLeave(t *testing.T) {
fp := "src2:" + strings.Repeat("a", 64)
at := time.Now().UTC()
js, ctx, _, _ := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) {
@@ -1464,4 +1465,12 @@ func TestIndexerKeepsWhatAWantAndARunningBuildName(t *testing.T) {
if got := members(2); !slices.Equal(got, []string{"n1.avahi", "n2.avahi"}) {
t.Fatalf("named holds %v for the fingerprint within 20 s of the start, want both members", got)
}
gone := at.Add(time.Minute)
removed, _ := json.Marshal(meshwants.WantedEvent{V: 1, Node: "n2", Module: "avahi", Fp: fp, Revision: 2, Removed: true, At: &gone})
if _, err := js.Publish(ctx, meshwants.SubjectWanted("n2", "avahi"), removed); err != nil {
t.Fatal(err)
}
if got := members(1); !slices.Equal(got, []string{"n1.avahi"}) {
t.Errorf("named holds %v after the want of n2 was said removed, want n1.avahi alone", got)
}
}
modules/mesh-delivery-indexer/internal/index/named.go
+24 -8
View File
@@ -66,7 +66,8 @@ type Named struct {
mu sync.Mutex
want map[string]namedAt
run map[string]namedAt
reads map[string]bool // the streams read whole at the start
gone map[string]time.Time // the want of a member removed (§693), when
reads map[string]bool // the streams read whole at the start
ready bool
}
@@ -75,7 +76,7 @@ func NewNamed(store Store, logf func(string, ...any)) *Named {
if logf == nil {
logf = func(string, ...any) {}
}
return &Named{Store: store, Logf: logf, want: map[string]namedAt{}, run: map[string]namedAt{}, reads: map[string]bool{}}
return &Named{Store: store, Logf: logf, want: map[string]namedAt{}, run: map[string]namedAt{}, gone: map[string]time.Time{}, reads: map[string]bool{}}
}
// Ready says the two streams were read whole and the sets reconciled with them: until then no entry object is removed, so a
@@ -86,7 +87,8 @@ func (n *Named) Ready() bool {
return n.ready
}
// HearWanted applies one wanted event, said on mesh.seat.mesh-wants.event.wanted.<node>.<module>: the want names its fingerprint.
// HearWanted applies one wanted event, said on mesh.seat.mesh-wants.event.wanted.<node>.<module>: the want names its fingerprint,
// or, with removed, names none and leaves the member out of every set whatever its node last said running (§693).
func (n *Named) HearWanted(ctx context.Context, data []byte) error {
var e meshwants.WantedEvent
if err := json.Unmarshal(data, &e); err != nil {
@@ -99,7 +101,7 @@ func (n *Named) HearWanted(ctx context.Context, data []byte) error {
if e.At != nil {
at = *e.At
}
return n.apply(ctx, NamedMember(e.Node, e.Module), true, e.Fp, at)
return n.apply(ctx, NamedMember(e.Node, e.Module), true, e.Fp, at, e.Removed)
}
// HearRunning applies one running event: the build it names; a stopped one, or one with no fingerprint, names none (design 47
@@ -116,14 +118,14 @@ func (n *Named) HearRunning(ctx context.Context, data []byte) error {
if r.Stopped {
fp = ""
}
return n.apply(ctx, NamedMember(r.Node, r.Module), false, fp, r.At)
return n.apply(ctx, NamedMember(r.Node, r.Module), false, fp, r.At, false)
}
// apply changes what one source names for the member and writes the sets it moves; one at a time, so the two sources never
// write one set from a stale view of each other.
func (n *Named) apply(ctx context.Context, member string, wanted bool, fp string, at time.Time) error {
func (n *Named) apply(ctx context.Context, member string, wanted bool, fp string, at time.Time, removed bool) error {
n.mu.Lock()
moved, err := n.applyLocked(ctx, member, wanted, fp, at)
moved, err := n.applyLocked(ctx, member, wanted, fp, at, removed)
n.mu.Unlock()
if moved && n.Changed != nil {
n.Changed(ctx)
@@ -131,7 +133,7 @@ func (n *Named) apply(ctx context.Context, member string, wanted bool, fp string
return err
}
func (n *Named) applyLocked(ctx context.Context, member string, wanted bool, fp string, at time.Time) (bool, error) {
func (n *Named) applyLocked(ctx context.Context, member string, wanted bool, fp string, at time.Time, removed bool) (bool, error) {
src := n.run
if wanted {
src = n.want
@@ -145,9 +147,23 @@ func (n *Named) applyLocked(ctx context.Context, member string, wanted bool, fp
touched = append(touched, old.fp)
}
switch {
case removed:
// The node left or the module was unplaced: the want goes, and with it what the node last said running.
if r, ok := n.run[member]; ok {
touched = append(touched, r.fp)
delete(n.run, member)
}
delete(n.want, member)
n.gone[member] = at
touched = append(touched, fp)
case !n.gone[member].IsZero() && !at.After(n.gone[member]):
return false, nil // an event older than the removal of the want it belongs to
case fp == "":
delete(src, member)
default:
if wanted {
delete(n.gone, member)
}
src[member] = namedAt{fp: fp, at: at}
touched = append(touched, fp)
}
modules/mesh-delivery-indexer/internal/index/wanted_names_test.go
+7 -1
View File
@@ -27,6 +27,7 @@ type namedEvent struct {
module string
fp string
min int
removed bool
stopped bool
}
@@ -36,7 +37,7 @@ func (e namedEvent) apply(t *testing.T, n *Named) {
var data []byte
var err error
if e.wanted {
data, err = json.Marshal(meshwants.WantedEvent{V: 1, Node: e.node, Module: e.module, Fp: e.fp, Revision: 1, At: &at})
data, err = json.Marshal(meshwants.WantedEvent{V: 1, Node: e.node, Module: e.module, Fp: e.fp, Revision: 1, Removed: e.removed, At: &at})
if err == nil {
err = n.HearWanted(context.Background(), data)
}
@@ -69,6 +70,11 @@ func TestNamedHoldsTheMembersWhoseWantOrRunningBuildNamesAFingerprint(t *testing
{wanted: true, node: "n1", module: "m", fp: b, min: 2}}, map[string][]string{a: {"n1.m"}, b: {"n1.m"}}},
{"a stopped running names nothing", []namedEvent{
{node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", stopped: true, min: 2}}, map[string][]string{a: nil}},
{"a want removed empties the member whatever its node last said running (§693)", []namedEvent{
{wanted: true, node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", fp: a, min: 1},
{wanted: true, node: "n1", module: "m", fp: a, min: 3, removed: true}}, map[string][]string{a: nil}},
{"a running older than the removal does not bring the member back", []namedEvent{
{wanted: true, node: "n1", module: "m", fp: a, min: 1, removed: true}, {node: "n1", module: "m", fp: a, min: 0}}, map[string][]string{a: nil}},
{"an event older than the one heard changes nothing", []namedEvent{
{wanted: true, node: "n1", module: "m", fp: b, min: 5}, {wanted: true, node: "n1", module: "m", fp: a, min: 4}}, map[string][]string{a: nil, b: {"n1.m"}}},
} {