novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

Read this node's placement from its node and module keys, or its placement.<node> before the holder converts it, so a write of the holder's is composed only once its node key lands (issue 774, research 072 row 37)

Browse Source
This commit is contained in:
jschoubben
2026-10-31 09:37:23 +01:00
parent bb8383b065
commit a4c0cee251
8 changed files with 176 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/link/watch.go
+8 -2
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/nats-io/nats.go"
@@ -23,7 +24,9 @@ type KVUpdate struct {
// 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.
// or the link does (the channel is then closed). A bucket the node's account may not read is an error. The pattern
// may be several, separated by spaces (WatchPatterns), watched as one: a node's placement in either form of its
// keys (research 072 row 37).
KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error)
}
@@ -41,7 +44,7 @@ func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan K
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
}
w, err := kv.Watch(pattern, nats.Context(ctx))
w, err := kv.WatchFiltered(strings.Fields(pattern), nats.Context(ctx))
if err != nil {
return nil, fmt.Errorf("watching %s in %s: %w", pattern, bucket, err)
}
@@ -296,3 +299,6 @@ func (q *Session) KVWatchAwait(ctx context.Context, bucket, pattern string, wait
}()
return out, nil
}
// WatchPatterns is several key patterns as one pattern KVWatch takes.
func WatchPatterns(patterns ...string) string { return strings.Join(patterns, " ") }
internal/selfcompose/selfcompose.go
+29 -12
View File
@@ -98,7 +98,10 @@ type Composer struct {
mu sync.Mutex
wants map[string]delivery.Want // by module, the node's own under agent.NodeModule
placements map[string]delivery.Placement
records map[string]delivery.NodeRecord
// placementKeys is the seat's state placement key by key in either form, each node's placement assembled from it
// (delivery.PlacementKeys; research 072 row 37): a write of the holder's under way is never composed half.
placementKeys *delivery.PlacementKeys
records map[string]delivery.NodeRecord
// inOwn is, by module, whether this node's have of it names no fingerprint of the new form (mesh-wireguard, a module
// at a build of the old form): the node runs it inside its own composition, and composes it there again (issue 1126).
inOwn map[string]bool
@@ -215,7 +218,7 @@ func New(env Env) *Composer {
if env.Say == nil {
env.Say = func(string) {}
}
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{},
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, placementKeys: delivery.NewPlacementKeys(),
records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{},
asked: make(chan chan Pass, 8)}
@@ -499,7 +502,9 @@ func (c *Composer) entry(bucket, key string) (v any, has, known bool) {
w, ok := c.wants[module]
return w.FP, ok, true
case delivery.PlacementBucket:
node, err := delivery.ParsePlacementKey(key)
// Any key of a node, its node key, a module key or its legacy key, stands for the node's whole placement: a
// module key written before its node key names it changes nothing composed.
node, _, _, err := delivery.ParsePlacementKey(key)
if err != nil {
return nil, false, true
}
@@ -551,18 +556,25 @@ func (c *Composer) takeWant(u agent.Update) {
}
func (c *Composer) takePlacement(u agent.Update) {
node, err := delivery.ParsePlacementKey(u.Key)
if err != nil {
if u.Caught {
return
}
c.mu.Lock()
defer c.mu.Unlock()
p, err := delivery.DecodePlacement(u.Value)
if u.Deleted || err != nil || p.Node != node {
delete(c.placements, node)
return
if c.placementKeys == nil {
c.placementKeys = delivery.NewPlacementKeys()
}
node, err := c.placementKeys.Put(u.Key, u.Value, u.Revision, u.Deleted || len(u.Value) == 0)
if node != "" {
if p, _, ok := c.placementKeys.Of(node); ok {
c.placements[node] = p
} else {
delete(c.placements, node)
}
}
c.mu.Unlock()
if node != "" && err != nil && c.env.Say != nil {
c.env.Say("the seat's placement of " + node + ": " + err.Error())
}
c.placements[node] = p
}
func (c *Composer) takeRecord(u agent.Update) {
@@ -1370,13 +1382,18 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
// watch endWhenPlaced keeps). A node that switched never goes back on a failed read, however long it lasts.
func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string,
wait func(try int) time.Duration, say func(string), switched bool) (bool, error) {
key, err := delivery.PlacementKey(node)
key, err := delivery.PlacementNodeKey(node)
if err != nil {
return false, err
}
legacy, _ := delivery.LegacyPlacementKey(node)
said := ""
for try := 1; ; try++ {
// Its node key, or, before the seat's holder converted it, its legacy key (research 072 row 37).
raw, err := get(ctx, delivery.PlacementBucket, key)
if err == nil && len(raw) == 0 {
raw, err = get(ctx, delivery.PlacementBucket, legacy)
}
if errors.Is(err, nats.ErrBucketNotFound) {
return false, nil // the seat has written no placement state yet: nothing places it, read and known
}
internal/selfcompose/selfcompose_test.go
+57 -4
View File
@@ -550,7 +550,50 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
// secrets. A key written again with what it held, or a want whose words moved and not its build, makes none.
worded, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true, Words: "lemurs is held at aaaaaaaaaaaa"})
elsewhere, _ := delivery.EncodeWant(delivery.Want{FP: delivery.SourcePrefix2 + strings.Repeat("e", 64), Self: true})
for _, row := range []struct {
// Research 072 row 37: the placement converted to one key per module and node moves nothing composed; a write in
// keys makes its pass at its node key, the commit, and not at a module key written before it nor at the mirror.
placeBase := delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}}
placeKeys := delivery.NewPlacementKeys()
keyRows := func(after delivery.Placement, what string) []struct {
name, bucket, key string
value []byte
pass bool
} {
writes, err := placeKeys.Writes("n1", &after)
if err != nil {
t.Fatal(err)
}
var rows []struct {
name, bucket, key string
value []byte
pass bool
}
for i, w := range writes {
_, _ = placeKeys.Put(w.Key, w.Value, uint64(i+1), w.Value == nil)
rows = append(rows, struct {
name, bucket, key string
value []byte
pass bool
}{what + ": " + w.Key, delivery.PlacementBucket, w.Key, w.Value, w.Commit && what != "converted"})
}
return rows
}
_, _ = placeKeys.Put("placement.n1", place, 1, false)
lemursMoved := placeBase
lemursMoved.Modules = []delivery.PlacedModule{{Module: "lemurs", Values: map[string]any{"greeting": "hoi"}}, {Module: "oldie"}}
var placementRows []struct {
name, bucket, key string
value []byte
pass bool
}
for _, step := range []struct {
what string
p delivery.Placement
}{{"converted", placeBase}, {"lemurs moved", lemursMoved}, {"lemurs back", placeBase}} {
placementRows = append(placementRows, keyRows(step.p, step.what)...)
}
for _, row := range append([]struct {
name, bucket, key string
value []byte
pass bool
@@ -562,7 +605,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
{"lemurs's module again", delivery.ModulesBucket, "lemurs", mod, false},
{"lemurs's want at another build", delivery.AssignerWantsBucket, "want.n1.lemurs", elsewhere, true},
{"lemurs's want back at its build", delivery.AssignerWantsBucket, "want.n1.lemurs", want, true},
} {
}, placementRows...) {
before := passes()
b.put(t, row.bucket, row.key, row.value)
if !row.pass {
@@ -690,7 +733,17 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
both, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}, {Module: "keeper"}}})
b.put(t, delivery.ModulesBucket, "keeper", kmod)
b.put(t, delivery.PlacementBucket, "placement.n1", both)
// n1 is held in keys per module and node since the rows above, which a legacy key no longer moves.
bothPlaced, _ := delivery.DecodePlacement(both)
bothWrites, err := placeKeys.Writes("n1", &bothPlaced)
if err != nil {
t.Fatal(err)
}
for _, w := range bothWrites {
if w.Value != nil {
b.put(t, delivery.PlacementBucket, w.Key, w.Value)
}
}
b.put(t, delivery.AssignerWantsBucket, "want.n1.keeper", want)
keeperKey, _ := delivery.ComposedKey("n1", "keeper", fp)
var handed string
@@ -1227,7 +1280,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
}{
"switched: placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3, switched: true},
"never switched: a timeout is not placed, said once": {answers: []error{fmt.Errorf("nats: timeout")}, placed: false, tries: 1},
"no key": {placed: false, tries: 1},
"no key": {placed: false, tries: 2}, // its node key, then its legacy key (research 072 row 37)
"no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true},
"switched: failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"),
fmt.Errorf("jetstream not ready")}, end: true, tries: 3, switched: true},