Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a313c3e843 | ||
|
|
e7f51caedb | ||
|
|
4b194215dc |
@@ -7,6 +7,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"maps"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
@@ -248,7 +249,7 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "web"}}})
|
||||
plkey, _ := delivery.PlacementKey("n1")
|
||||
plkey, _ := delivery.LegacyPlacementKey("n1") // the form before, read beside the new (research 072 row 37)
|
||||
if _, err := placement.Put(pb.ctx, plkey, pl); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -470,7 +471,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pkey, _ := delivery.PlacementKey(node)
|
||||
pkey, _ := delivery.LegacyPlacementKey(node)
|
||||
b.putJSON(t, delivery.PlacementBucket, pkey, pl)
|
||||
}
|
||||
if w := b.wantNow(t, "ace", "web", "ace's want once placed: the build alone, composed by the node", func(w delivery.Want) bool { return w.Self }); w.Composed != "" || w.FP != fNew {
|
||||
@@ -1092,24 +1093,52 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
}
|
||||
b.putJSON(t, delivery.ModulesBucket, "web", []byte(`{"v":1}`))
|
||||
b.putJSON(t, delivery.ModulesBucket, "shell", []byte(`{"v":1}`))
|
||||
place := func(node string, modules ...string) uint64 {
|
||||
// place writes a node's placement as another writer would, in its node and module keys (research 072 row 37), and
|
||||
// answers its node key's revision; legacy writes it in the form before, placement.<node>, alone.
|
||||
place := func(node string, legacy bool, modules ...string) uint64 {
|
||||
p := delivery.Placement{Node: node, Modules: []delivery.PlacedModule{}}
|
||||
for _, m := range modules {
|
||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
raw, _ := delivery.EncodePlacement(p)
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
rev, err := b.buckets[delivery.PlacementBucket].Put(b.ctx, key, raw)
|
||||
values, _ := delivery.PlacementValues(p)
|
||||
key, _ := delivery.PlacementNodeKey(node)
|
||||
if legacy {
|
||||
key, _ = delivery.LegacyPlacementKey(node)
|
||||
values = map[string][]byte{key: values[key]}
|
||||
}
|
||||
for _, k := range slices.Sorted(maps.Keys(values)) {
|
||||
if k == key {
|
||||
continue
|
||||
}
|
||||
if _, err := b.buckets[delivery.PlacementBucket].Put(b.ctx, k, values[k]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
rev, err := b.buckets[delivery.PlacementBucket].Put(b.ctx, key, values[key])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return rev
|
||||
}
|
||||
anchorRev, laptopRev := place("anchor", "web"), place("laptop", "shell")
|
||||
place("anchor", true, "web")
|
||||
laptopRev := place("laptop", false, "shell")
|
||||
bound := judge.AnswerBound
|
||||
t.Cleanup(func() { judge.AnswerBound = bound })
|
||||
judge.AnswerBound = 2 * time.Second
|
||||
saysWhatItKeeps(t, b, b.start(t))
|
||||
// The holder converts anchor, held in the form before alone, to its node and module keys once at its start.
|
||||
var anchorRev uint64
|
||||
for deadline := time.Now().Add(15 * time.Second); anchorRev == 0; time.Sleep(50 * time.Millisecond) {
|
||||
if e, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, "anchor.node"); err == nil {
|
||||
anchorRev = e.Revision()
|
||||
if _, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, "anchor.web"); err != nil {
|
||||
t.Fatalf("anchor converted without its module key: %v", err)
|
||||
}
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("anchor in the form before was not converted")
|
||||
}
|
||||
}
|
||||
verbs := newSeatTools(bedStore{b}, func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"ok":true}`), nil })
|
||||
propose := func() delivery.PlacementProposal {
|
||||
t.Helper()
|
||||
@@ -1168,13 +1197,20 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
t.Fatal("no placement-changed heard")
|
||||
}
|
||||
for node, want := range map[string][]string{"anchor": {}, "laptop": {"shell", "web"}} {
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
e, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
// The node key names the modules, each module key carries its own, and the mirror of the form before follows.
|
||||
values := map[string][]byte{}
|
||||
for _, key := range []string{node + ".node", "placement." + node, node + ".web", node + ".shell"} {
|
||||
if e, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); err == nil {
|
||||
values[key] = e.Value()
|
||||
}
|
||||
}
|
||||
if got, err := delivery.DecodePlacement(e.Value()); err != nil || !slices.Equal(got.Assigned(), want) {
|
||||
t.Errorf("%s = %v, %v, want %v", node, got.Assigned(), err, want)
|
||||
got, err := delivery.PlacementsOf(values)
|
||||
if err != nil || !slices.Equal(got[node].Assigned(), want) {
|
||||
t.Errorf("%s = %v, %v, want %v", node, got[node].Assigned(), err, want)
|
||||
}
|
||||
mirror, _ := delivery.DecodePlacement(values["placement."+node])
|
||||
if !slices.Equal(mirror.Assigned(), want) {
|
||||
t.Errorf("the mirror placement.%s = %v, want %v", node, mirror.Assigned(), want)
|
||||
}
|
||||
}
|
||||
if b.hasKey(delivery.PlacementProposalsBucket, p.ID) {
|
||||
@@ -1187,7 +1223,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
moved := q.(proposed).Proposal
|
||||
place("laptop", "shell", "web") // another writer moved the key
|
||||
place("laptop", false, "shell", "web") // another writer moved the key
|
||||
relaid := proposalNow(moved.ID, "laid again", func(p delivery.PlacementProposal) bool {
|
||||
return strings.Contains(p.Words, "laid again") && len(p.Must) > 0
|
||||
})
|
||||
@@ -1200,7 +1236,7 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
t.Fatal("a declined proposal was not ended")
|
||||
}
|
||||
}
|
||||
key, _ := delivery.PlacementKey("laptop")
|
||||
key, _ := delivery.PlacementNodeKey("laptop")
|
||||
if e, _ := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); e == nil || !strings.Contains(string(e.Value()), "web") {
|
||||
t.Error("a declined proposal wrote its key")
|
||||
}
|
||||
|
||||
@@ -69,6 +69,8 @@ type passer struct {
|
||||
seat *seatActs
|
||||
publish publisher
|
||||
saidProposed map[string]bool
|
||||
// convertedAll says every node in the form before was converted to its node and module keys (convertPlacement).
|
||||
convertedAll bool
|
||||
// saidWaits is the last said of the seat's wait for its states (sayWaits).
|
||||
saidWaits string
|
||||
intended map[string]map[string]map[string]bool
|
||||
@@ -1219,16 +1221,7 @@ func (p *passer) seatHolders(ctx context.Context) {
|
||||
if kv == nil {
|
||||
return
|
||||
}
|
||||
placements := map[string]delivery.Placement{}
|
||||
for key, value := range p.w.read(delivery.PlacementBucket) {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if pl, err := delivery.DecodePlacement(value); err == nil && pl.Node == node {
|
||||
placements[node] = pl
|
||||
}
|
||||
}
|
||||
placements := p.w.placements()
|
||||
if len(placements) == 0 {
|
||||
return // no placement read yet: nothing is known of who holds what, and nothing is deleted
|
||||
}
|
||||
|
||||
@@ -844,12 +844,15 @@ func aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) {
|
||||
for _, m := range modules {
|
||||
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
raw, err := delivery.EncodePlacement(pl)
|
||||
// One node key and a key per module (research 072 row 37); a module taken off leaves its key, which the node
|
||||
// key no longer names.
|
||||
values, err := delivery.PlacementValues(pl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
p.w.put(delivery.PlacementBucket, key, raw)
|
||||
for key, raw := range values {
|
||||
p.w.put(delivery.PlacementBucket, key, raw)
|
||||
}
|
||||
}
|
||||
assigned("g14", "a")
|
||||
assigned("g14", "b")
|
||||
@@ -1132,7 +1135,7 @@ func TestARefusedCallClearsWhenTheSourcesAgree(t *testing.T) {
|
||||
p.seat = newSeatActs(bus, func() { woke <- struct{}{} }, func(string, ...any) {}, func() bool { return false })
|
||||
// The placement pins db on laptop; the controller does not.
|
||||
raw, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "laptop", Pins: map[string]delivery.Provider{"db": store}})
|
||||
key, _ := delivery.PlacementKey("laptop")
|
||||
key, _ := delivery.LegacyPlacementKey("laptop")
|
||||
p.w.put(delivery.PlacementBucket, key, raw)
|
||||
// pass is a pass, then one more for each background read it started.
|
||||
pass := func(differ []assign.Pair) {
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"reflect"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -416,19 +417,10 @@ func (p *passer) sayWaits(unbound []string) {
|
||||
p.said[c.Key], p.saidWaits = true, state
|
||||
}
|
||||
|
||||
// seatPlacements is every node's placement key and its revision, as the watch holds them.
|
||||
// seatPlacements is every node's placement and the revision of its node key (its legacy key's for a node not yet
|
||||
// converted), as the watch holds them: the revision a proposal is laid over (§2, research 072 row 37).
|
||||
func (p *passer) seatPlacements() (map[string]delivery.Placement, map[string]uint64) {
|
||||
out, revs := map[string]delivery.Placement{}, map[string]uint64{}
|
||||
for key, value := range p.w.read(delivery.PlacementBucket) {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if pl, err := delivery.DecodePlacement(value); err == nil && pl.Node == node {
|
||||
out[node], revs[node] = pl, p.w.metaOf(delivery.PlacementBucket, key).revision
|
||||
}
|
||||
}
|
||||
return out, revs
|
||||
return p.w.placement.AllRevisions()
|
||||
}
|
||||
|
||||
// candidatesOf is a proposal's candidates by node, with their keys' revisions and the newest write.
|
||||
@@ -482,6 +474,7 @@ func (p *passer) placementSeat(ctx context.Context) {
|
||||
if p.saidProposed == nil {
|
||||
p.saidProposed, p.intended = map[string]bool{}, map[string]map[string]map[string]bool{}
|
||||
}
|
||||
p.convertPlacement(ctx)
|
||||
current, revs := p.seatPlacements()
|
||||
ids := make([]string, 0)
|
||||
raws := p.w.read(delivery.PlacementProposalsBucket)
|
||||
@@ -815,27 +808,12 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal,
|
||||
p.logf("the proposal %s cannot be written: %v", prop.ID, err)
|
||||
return
|
||||
}
|
||||
kv := p.kvs[delivery.PlacementBucket]
|
||||
var changes []delivery.PlacementChange
|
||||
writtenLaid := map[string]*delivery.Placement{}
|
||||
after := maps.Clone(current)
|
||||
afterRevs := maps.Clone(revs)
|
||||
for _, w := range writes {
|
||||
var rev2 uint64
|
||||
var err error
|
||||
switch {
|
||||
case w.Value == nil:
|
||||
err = kv.Delete(ctx, w.Key, jetstream.LastRevision(w.Base))
|
||||
default:
|
||||
raw, encErr := delivery.EncodePlacement(*w.Value)
|
||||
if encErr != nil {
|
||||
err = encErr
|
||||
} else if w.Base == 0 {
|
||||
rev2, err = kv.Create(ctx, w.Key, raw)
|
||||
} else {
|
||||
rev2, err = kv.Update(ctx, w.Key, raw, w.Base)
|
||||
}
|
||||
}
|
||||
rev2, err := p.writeNode(ctx, w.Node, w.Value)
|
||||
if err != nil {
|
||||
// Another writer came first (a hand-back): what is written stands and is followed through; the rest is laid
|
||||
// again over the keys as this write left them.
|
||||
@@ -902,11 +880,17 @@ func (p *passer) handBack(ctx context.Context, prop delivery.PlacementProposal,
|
||||
return
|
||||
}
|
||||
answer, _ := p.w.answer(akey)
|
||||
current, _ := p.seatPlacements()
|
||||
current, held := p.seatPlacements()
|
||||
var changes []delivery.PlacementChange
|
||||
for _, node := range prop.Nodes() {
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
if err := p.kvs[delivery.PlacementBucket].Delete(ctx, key, jetstream.LastRevision(prop.Base[node])); err != nil {
|
||||
key, _ := delivery.PlacementNodeKey(node)
|
||||
err := error(nil)
|
||||
if held[node] != prop.Base[node] {
|
||||
err = fmt.Errorf("its placement is at revision %d, the hand-back read %d", held[node], prop.Base[node])
|
||||
} else {
|
||||
_, err = p.writeNode(ctx, node, nil)
|
||||
}
|
||||
if err != nil {
|
||||
p.sayChanged(prop, answer, changes)
|
||||
p.endProposal(ctx, prop, fmt.Sprintf("%s moved since the hand-back was asked (%v): hand back again", key, err))
|
||||
return
|
||||
@@ -1026,3 +1010,73 @@ func (p *passer) sayEvent(event string, body delivery.PlacementEvent, id string)
|
||||
p.logf("%s of %s is not said: %v", event, body.Proposal.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// writeNode writes one node's placement after (nil: none) as its keys (mesh-sdk delivery.PlacementKeys.Writes; research
|
||||
// 072 row 37): the module keys that change, each carrying the value it replaces, then the node key, the commit, then the
|
||||
// module keys no longer placed. It answers the node key's revision after (0: deleted). A write that fails up to the
|
||||
// commit ends it with nothing changed for any reader, the module keys written before it read at their previous values;
|
||||
// a delete after the commit that fails leaves a module key the node key does not name, which places nothing and is
|
||||
// deleted at the node's next write. Each key written is taken into the world at once, so the next write of the pass
|
||||
// reads it.
|
||||
func (p *passer) writeNode(ctx context.Context, node string, after *delivery.Placement) (uint64, error) {
|
||||
writes, err := p.w.placement.Writes(node, after)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
kv := p.kvs[delivery.PlacementBucket]
|
||||
committed, rev := false, uint64(0)
|
||||
for _, w := range writes {
|
||||
got, err := writeKey(ctx, kv, w)
|
||||
if err != nil {
|
||||
if committed {
|
||||
p.logf("placement %s is not deleted after %s's write (%v): it places nothing, and goes at the node's next write", w.Key, node, err)
|
||||
continue
|
||||
}
|
||||
return 0, fmt.Errorf("placement %s at revision %d: %w", w.Key, w.Base, err)
|
||||
}
|
||||
p.w.putEntry(delivery.PlacementBucket, w.Key, w.Value, got, p.clock())
|
||||
if w.Commit {
|
||||
committed, rev = true, got
|
||||
}
|
||||
}
|
||||
return rev, nil
|
||||
}
|
||||
|
||||
// writeKey makes one write of a placement key by compare-and-set and answers the key's revision after (0: deleted).
|
||||
func writeKey(ctx context.Context, kv jetstream.KeyValue, w delivery.PlacementKeyWrite) (uint64, error) {
|
||||
switch {
|
||||
case w.Value == nil:
|
||||
return 0, kv.Delete(ctx, w.Key, jetstream.LastRevision(w.Base))
|
||||
case w.Base == 0:
|
||||
return kv.Create(ctx, w.Key, w.Value)
|
||||
default:
|
||||
return kv.Update(ctx, w.Key, w.Value, w.Base)
|
||||
}
|
||||
}
|
||||
|
||||
// convertPlacement converts every node the state holds only in the form before, `placement.<node>`, into its node key
|
||||
// and module keys (research 072 row 37), once: each node converted is said, and a node whose conversion fails is said
|
||||
// and tried again at the next pass. The legacy key is left for the readers not yet delivered; a node key wins over it
|
||||
// for every reader that reads both, and it is deleted with the node key when the node's placement is.
|
||||
func (p *passer) convertPlacement(ctx context.Context) {
|
||||
if p.convertedAll {
|
||||
return
|
||||
}
|
||||
migrations, err := p.w.placement.Migrations()
|
||||
if err != nil {
|
||||
p.logf("placement: some nodes in the form before cannot be converted, and stay read from it: %v", err)
|
||||
}
|
||||
failed := false
|
||||
for _, node := range slices.Sorted(maps.Keys(migrations)) {
|
||||
before, rev, _ := p.w.placement.Of(node)
|
||||
got, err := p.writeNode(ctx, node, &before)
|
||||
if err != nil {
|
||||
failed = true
|
||||
p.logf("placement: %s is not converted from placement.%s (revision %d) to its node and module keys, and is tried again at the next pass: %v", node, node, rev, err)
|
||||
continue
|
||||
}
|
||||
p.logf("placement: %s converted from placement.%s (revision %d) to %s.node (revision %d) and %d module keys (research 072 row 37); placement.%s stays for readers not yet delivered",
|
||||
node, node, rev, node, got, len(before.Modules), node)
|
||||
}
|
||||
p.convertedAll = !failed && err == nil
|
||||
}
|
||||
|
||||
@@ -60,30 +60,34 @@ type keyed struct {
|
||||
Revision uint64 `json:"revision"`
|
||||
}
|
||||
|
||||
// placements reads every node's placement and its key's revision.
|
||||
// placements reads every node's placement and the revision of its node key (its legacy key's before it is converted),
|
||||
// assembled from its keys in either form (delivery.PlacementKeys; research 072 row 37).
|
||||
func (s *seatTools) placements() (map[string]delivery.Placement, map[string]uint64, error) {
|
||||
keys, err := s.st.keys(delivery.PlacementState)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("placement cannot be listed: %w", err)
|
||||
}
|
||||
out, revs := map[string]delivery.Placement{}, map[string]uint64{}
|
||||
sort.Strings(keys)
|
||||
held := delivery.NewPlacementKeys()
|
||||
var bad []string
|
||||
for _, key := range keys {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
if _, _, _, err := delivery.ParsePlacementKey(key); err != nil {
|
||||
continue
|
||||
}
|
||||
raw, rev, found, err := s.st.get(delivery.PlacementState, key)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("placement of %s cannot be read: %w", node, err)
|
||||
return nil, nil, fmt.Errorf("placement key %s cannot be read: %w", key, err)
|
||||
}
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
p, err := delivery.DecodePlacement(raw)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("placement of %s cannot be read: %w", node, err)
|
||||
if _, err := held.Put(key, raw, rev, false); err != nil {
|
||||
bad = append(bad, err.Error())
|
||||
}
|
||||
out[node], revs[node] = p, rev
|
||||
}
|
||||
out, revs := held.AllRevisions()
|
||||
if len(bad) > 0 && len(out) == 0 {
|
||||
return nil, nil, fmt.Errorf("placement cannot be read: %s", strings.Join(bad, "; "))
|
||||
}
|
||||
return out, revs, nil
|
||||
}
|
||||
@@ -413,26 +417,20 @@ func (s *seatTools) doPlacement(args map[string]any) (any, error) {
|
||||
sort.Strings(nodes)
|
||||
}
|
||||
for _, n := range nodes {
|
||||
key, err := delivery.PlacementKey(n)
|
||||
revisions, err := s.nodeHistory(n)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
revisions, err := s.st.history(delivery.PlacementState, key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the history of %s cannot be read: %w", key, err)
|
||||
}
|
||||
var before delivery.Placement
|
||||
var before sketch
|
||||
for _, r := range revisions {
|
||||
e := historyEntry{Node: n, Revision: r.Revision, At: r.At}
|
||||
after := delivery.Placement{}
|
||||
after := sketch{}
|
||||
if r.Op == "" || r.Op == "put" || r.Op == "PUT" {
|
||||
if p, err := delivery.DecodePlacement(r.Value); err == nil {
|
||||
after = p
|
||||
}
|
||||
after = sketchOf(r.Key, r.Value)
|
||||
} else {
|
||||
e.Deleted = true
|
||||
}
|
||||
e.Change = delivery.ChangeOf(before, after)
|
||||
e.Change = after.changeFrom(n, before)
|
||||
e.Change.Before, e.Change.After = 0, r.Revision
|
||||
before = after
|
||||
if module != "" && !strings.Contains(" "+strings.Join(append(append(e.Change.Added, e.Change.Removed...), e.Change.Changed...), " ")+" ", " "+module+" ") {
|
||||
@@ -571,3 +569,100 @@ func (s *seatTools) seats(node string) (any, error) {
|
||||
}
|
||||
return map[string]any{"seats": out, "holds": len(keys)}, nil
|
||||
}
|
||||
|
||||
// keyedRevision is one revision of a node's placement key, the key it is of beside it.
|
||||
type keyedRevision struct {
|
||||
stdio.StateRevision
|
||||
Key string
|
||||
}
|
||||
|
||||
// nodeHistory is the revisions of a node's placement, in the order the bus wrote them: its legacy key's
|
||||
// `placement.<node>` before it was converted, then its node key's `<node>.node`, whose every revision is one write of
|
||||
// the node (research 072 row 37: the node key is a write's commit, so its history is the node's placement history).
|
||||
func (s *seatTools) nodeHistory(node string) ([]keyedRevision, error) {
|
||||
legacy, err := delivery.LegacyPlacementKey(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
nodeKey, err := delivery.PlacementNodeKey(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []keyedRevision
|
||||
for _, key := range []string{legacy, nodeKey} {
|
||||
revisions, err := s.st.history(delivery.PlacementState, key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the history of %s cannot be read: %w", key, err)
|
||||
}
|
||||
for _, r := range revisions {
|
||||
out = append(out, keyedRevision{StateRevision: r, Key: key})
|
||||
}
|
||||
}
|
||||
sort.SliceStable(out, func(i, j int) bool { return out[i].Revision < out[j].Revision })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// sketch is a revision of a node's placement as far as its history says a change: each module's digest, and its pins
|
||||
// and holdings, which change every module (delivery.ChangeOf).
|
||||
type sketch struct {
|
||||
modules []delivery.PlacementRef
|
||||
nodeWide string
|
||||
}
|
||||
|
||||
// sketchOf reads one revision of a node key or a legacy key; a value that does not read places nothing.
|
||||
func sketchOf(key string, value []byte) sketch {
|
||||
_, _, kind, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
return sketch{}
|
||||
}
|
||||
wide := func(pins map[string]delivery.Provider, holds []delivery.Holding) string {
|
||||
raw, _ := json.Marshal([]any{pins, holds})
|
||||
return string(raw)
|
||||
}
|
||||
if kind == delivery.PlacementNodeKind {
|
||||
n, err := delivery.DecodePlacementNode(value)
|
||||
if err != nil {
|
||||
return sketch{}
|
||||
}
|
||||
return sketch{modules: n.Modules, nodeWide: wide(n.Pins, n.Holds)}
|
||||
}
|
||||
p, err := delivery.DecodePlacement(value)
|
||||
if err != nil {
|
||||
return sketch{}
|
||||
}
|
||||
out := sketch{nodeWide: wide(p.Pins, p.Holds)}
|
||||
for _, m := range p.Modules {
|
||||
out.modules = append(out.modules, delivery.PlacementRef{Module: m.Module, Digest: m.Digest()})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// changeFrom is what changed from before to s on node, as delivery.ChangeOf says it.
|
||||
func (s sketch) changeFrom(node string, before sketch) delivery.PlacementChange {
|
||||
c := delivery.PlacementChange{Node: node}
|
||||
was := map[string]string{}
|
||||
for _, r := range before.modules {
|
||||
was[r.Module] = r.Digest
|
||||
}
|
||||
now := map[string]bool{}
|
||||
nodeWide := before.nodeWide != s.nodeWide && len(before.modules) > 0
|
||||
for _, r := range s.modules {
|
||||
now[r.Module] = true
|
||||
d, ok := was[r.Module]
|
||||
switch {
|
||||
case !ok:
|
||||
c.Added = append(c.Added, r.Module)
|
||||
case nodeWide || d != r.Digest:
|
||||
c.Changed = append(c.Changed, r.Module)
|
||||
}
|
||||
}
|
||||
for _, r := range before.modules {
|
||||
if !now[r.Module] {
|
||||
c.Removed = append(c.Removed, r.Module)
|
||||
}
|
||||
}
|
||||
sort.Strings(c.Added)
|
||||
sort.Strings(c.Removed)
|
||||
sort.Strings(c.Changed)
|
||||
return c
|
||||
}
|
||||
|
||||
@@ -3,6 +3,8 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
@@ -14,6 +16,7 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
@@ -168,7 +171,7 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
Policy: catalogue.PolicyRoll, PolicyFrom: catalogue.FromDefault})
|
||||
w.putEntry(delivery.ControllerComposedBucket, ckey, cv, 1, time.Now())
|
||||
pl, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "ace", Modules: []delivery.PlacedModule{{Module: "web"}}})
|
||||
pkey, _ := delivery.PlacementKey("ace")
|
||||
pkey, _ := delivery.LegacyPlacementKey("ace")
|
||||
w.putEntry(delivery.PlacementBucket, pkey, pl, 1, time.Now())
|
||||
for node, digest := range map[string]string{"ace": "sha256:" + strings.Repeat("a", 64), "g14": ""} {
|
||||
hn, _ := delivery.EncodeHaveNode(delivery.HaveNode{Composed: digest})
|
||||
@@ -194,7 +197,7 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
t.Fatalf("findings = %v", findings)
|
||||
}
|
||||
pl2, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}})
|
||||
pkey2, _ := delivery.PlacementKey("g14")
|
||||
pkey2, _ := delivery.LegacyPlacementKey("g14")
|
||||
w.putEntry(delivery.PlacementBucket, pkey2, pl2, 1, time.Now())
|
||||
p.unplacedFindings()
|
||||
if len(findings) != 0 {
|
||||
@@ -210,7 +213,7 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
k, _ := delivery.PlacementKey(node)
|
||||
k, _ := delivery.LegacyPlacementKey(node)
|
||||
p.w.putEntry(delivery.PlacementBucket, k, pl, 1, time.Now())
|
||||
}
|
||||
place("ace", delivery.Holding{Claim: "mesh-broker", Scope: "mesh", Module: "nats"}, delivery.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
|
||||
@@ -235,6 +238,68 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
t.Errorf("seats on ace = %+v, %v", out, err)
|
||||
}
|
||||
})
|
||||
t.Run("research 072 row 37: the form before is converted once, a node is written as its keys, and history reads both", func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
kv := &revKV{values: map[string][]byte{}, revs: map[string]uint64{}}
|
||||
var logs []string
|
||||
p := &passer{w: newWorld(), kvs: buckets{delivery.PlacementBucket: kv}, now: time.Now,
|
||||
logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) }}
|
||||
put := func(key string, value []byte, rev uint64) {
|
||||
p.w.putEntry(delivery.PlacementBucket, key, value, rev, time.Now())
|
||||
}
|
||||
was := delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "web"}}}
|
||||
raw, _ := delivery.EncodePlacement(was)
|
||||
put("placement.anchor", raw, kv.write("placement.anchor", raw))
|
||||
p.convertPlacement(ctx)
|
||||
if strings.Join(kv.log, " ") != "create anchor.web create anchor.node" || !p.convertedAll ||
|
||||
!strings.Contains(strings.Join(logs, "\n"), "anchor converted from placement.anchor") {
|
||||
t.Fatalf("the conversion wrote %v, said %v", kv.log, logs)
|
||||
}
|
||||
p.convertPlacement(ctx)
|
||||
if len(kv.log) != 2 {
|
||||
t.Errorf("a converted node is converted again: %v", kv.log)
|
||||
}
|
||||
// A write: the module keys, the node key (its commit), the mirror, then the module key no longer placed.
|
||||
after := delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "shell", Ports: map[string]int{"22": 2222}}}}
|
||||
kv.log = nil
|
||||
rev, err := p.writeNode(ctx, "anchor", &after)
|
||||
if err != nil || strings.Join(kv.log, " ") != "create anchor.shell update anchor.node update placement.anchor delete anchor.web" || rev != kv.revs["anchor.node"] {
|
||||
t.Fatalf("the write = %v, %d, %v", kv.log, rev, err)
|
||||
}
|
||||
if got, _ := p.seatPlacements(); !delivery.SamePlacement(got["anchor"], after) {
|
||||
t.Errorf("the write is not read back at once: %+v", got)
|
||||
}
|
||||
// A commit refused leaves every reader on what stood: the module key written before it carries the value it replaced.
|
||||
kv.refuse, kv.log = "anchor.node", nil
|
||||
moved := delivery.Placement{Node: "anchor", Modules: []delivery.PlacedModule{{Module: "shell", Ports: map[string]int{"22": 2223}}}}
|
||||
if _, err := p.writeNode(ctx, "anchor", &moved); err == nil || strings.Join(kv.log, " ") != "update anchor.shell" {
|
||||
t.Fatalf("a refused commit = %v, %v", kv.log, err)
|
||||
}
|
||||
fresh, err := delivery.PlacementsOf(kv.values)
|
||||
if err != nil || !delivery.SamePlacement(fresh["anchor"], after) {
|
||||
t.Errorf("a reader of the keys after a refused commit reads %+v, %v", fresh["anchor"], err)
|
||||
}
|
||||
kv.refuse = ""
|
||||
// The verb's history reads the form before, then the node key.
|
||||
st := memStore{delivery.PlacementState: &memKV{values: kv.values}}
|
||||
hs := newSeatTools(st, nil)
|
||||
hs.st = historyStore{memStore: st, revisions: map[string][]stdio.StateRevision{
|
||||
"placement.anchor": {{Key: "placement.anchor", Value: raw, Revision: 1, Op: "put"}},
|
||||
"anchor.node": {{Key: "anchor.node", Value: kv.values["anchor.node"], Revision: 5, Op: "put"}, {Key: "anchor.node", Revision: 9, Op: "del"}},
|
||||
}}
|
||||
pa, err := hs.doPlacement(map[string]any{"node": "anchor", "history": "true"})
|
||||
a, _ := pa.(placementAnswer)
|
||||
if err != nil || len(a.Changes) != 3 || !slices.Equal(a.Changes[0].Change.Added, []string{"web"}) ||
|
||||
!slices.Equal(a.Changes[1].Change.Added, []string{"shell"}) || !slices.Equal(a.Changes[1].Change.Removed, []string{"web"}) ||
|
||||
!a.Changes[2].Deleted || !slices.Equal(a.Changes[2].Change.Removed, []string{"shell"}) {
|
||||
t.Errorf("history = %+v, %v", pa, err)
|
||||
}
|
||||
// No placement: the mirror first, then the node key, then the module keys.
|
||||
kv.log = nil
|
||||
if _, err := p.writeNode(ctx, "anchor", nil); err != nil || strings.Join(kv.log, " ") != "delete placement.anchor delete anchor.node delete anchor.shell" {
|
||||
t.Errorf("the delete = %v, %v", kv.log, err)
|
||||
}
|
||||
})
|
||||
t.Run("1954/D1: a failed provision is the assigner's finding until it is provided", func(t *testing.T) {
|
||||
findings := map[string][]byte{}
|
||||
p := &passer{w: newWorld(), logf: func(string, ...any) {}, kvs: buckets{provisions.ProvisionsBucket: nil}, findings: assign.Findings{
|
||||
@@ -412,3 +477,56 @@ func (c *countingKV) Put(ctx context.Context, key string, value []byte) (uint64,
|
||||
*c.writes++
|
||||
return c.memKV.Put(ctx, key, value)
|
||||
}
|
||||
|
||||
// revKV is a bucket with real revisions, refusing a write at a revision that is not the key's, and a write of refuse;
|
||||
// log is every write in order.
|
||||
type revKV struct {
|
||||
jetstream.KeyValue
|
||||
rev uint64
|
||||
values map[string][]byte
|
||||
revs map[string]uint64
|
||||
log []string
|
||||
refuse string
|
||||
}
|
||||
|
||||
func (k *revKV) write(key string, value []byte) uint64 {
|
||||
k.rev++
|
||||
k.values[key], k.revs[key] = value, k.rev
|
||||
return k.rev
|
||||
}
|
||||
|
||||
func (k *revKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) {
|
||||
if _, ok := k.values[key]; ok || key == k.refuse {
|
||||
return 0, errors.New("key exists")
|
||||
}
|
||||
k.log = append(k.log, "create "+key)
|
||||
return k.write(key, value), nil
|
||||
}
|
||||
|
||||
func (k *revKV) Update(_ context.Context, key string, value []byte, revision uint64) (uint64, error) {
|
||||
if k.revs[key] != revision || key == k.refuse {
|
||||
return 0, errors.New("wrong last sequence")
|
||||
}
|
||||
k.log = append(k.log, "update "+key)
|
||||
return k.write(key, value), nil
|
||||
}
|
||||
|
||||
func (k *revKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
|
||||
if key == k.refuse {
|
||||
return errors.New("wrong last sequence")
|
||||
}
|
||||
k.log = append(k.log, "delete "+key)
|
||||
delete(k.values, key)
|
||||
delete(k.revs, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// historyStore is a memStore whose history is given.
|
||||
type historyStore struct {
|
||||
memStore
|
||||
revisions map[string][]stdio.StateRevision
|
||||
}
|
||||
|
||||
func (s historyStore) history(_, key string) ([]stdio.StateRevision, error) {
|
||||
return s.revisions[key], nil
|
||||
}
|
||||
|
||||
@@ -54,6 +54,9 @@ type world struct {
|
||||
meta map[string]entryMeta
|
||||
// trunkBuilt is, by module, its newest done trunk build in the builder's log (putBuilt).
|
||||
trunkBuilt map[string]delivery.BuildLogEntry
|
||||
// placement is the seat's state placement assembled node by node from its keys in either form (mesh-sdk
|
||||
// delivery.PlacementKeys; research 072 row 37): the one reader of its keys in the assigner.
|
||||
placement *delivery.PlacementKeys
|
||||
}
|
||||
|
||||
// entryMeta is a value's revision in its bucket and when it was written.
|
||||
@@ -71,7 +74,7 @@ type parsed struct {
|
||||
|
||||
func newWorld() *world {
|
||||
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{},
|
||||
meta: map[string]entryMeta{}, trunkBuilt: map[string]delivery.BuildLogEntry{}}
|
||||
meta: map[string]entryMeta{}, trunkBuilt: map[string]delivery.BuildLogEntry{}, placement: delivery.NewPlacementKeys()}
|
||||
}
|
||||
|
||||
// putEntry is put, keeping the value's revision and the time it was written.
|
||||
@@ -84,6 +87,9 @@ func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at t
|
||||
return
|
||||
}
|
||||
w.meta[bucket+"|"+key] = entryMeta{revision: revision, at: at}
|
||||
if bucket == delivery.PlacementBucket {
|
||||
_, _ = w.placement.Put(key, value, revision, false) // put took it already; this is its revision
|
||||
}
|
||||
}
|
||||
|
||||
// metaOf is the revision and time of the value a bucket holds at key, as the watch said them.
|
||||
@@ -103,10 +109,17 @@ func (w *world) put(bucket, key string, value []byte) {
|
||||
w.changes[bucket]++ // the bucket's generation (bucketGen), a delete included
|
||||
if value == nil {
|
||||
delete(w.values[bucket], key)
|
||||
if bucket == delivery.PlacementBucket {
|
||||
_, _ = w.placement.Put(key, nil, 0, true)
|
||||
}
|
||||
return
|
||||
}
|
||||
w.values[bucket][key] = value
|
||||
w.changes[bucket+"|"+key]++
|
||||
if bucket == delivery.PlacementBucket {
|
||||
// A key or value that does not read places nothing new on its node; the indexer's read tool says it.
|
||||
_, _ = w.placement.Put(key, value, 0, false)
|
||||
}
|
||||
if bucket == delivery.HaveBucket {
|
||||
if h, err := delivery.DecodeHave(value); err == nil && h.FP != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK {
|
||||
list := slices.DeleteFunc(w.healthy[key], func(fp string) bool { return fp == h.FP })
|
||||
@@ -235,19 +248,16 @@ func (w *world) handedAway() bool {
|
||||
// that does not read places nothing on its node and is said by the indexer's own read tool.
|
||||
func (w *world) placed() (map[string][]string, bool) {
|
||||
out := map[string][]string{}
|
||||
values := w.read(delivery.PlacementBucket)
|
||||
for key, value := range values {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
p, err := delivery.DecodePlacement(value)
|
||||
if err != nil || p.Node != node {
|
||||
continue
|
||||
}
|
||||
for node, p := range w.placements() {
|
||||
out[node] = p.Assigned()
|
||||
}
|
||||
return out, len(values) > 0
|
||||
return out, len(w.read(delivery.PlacementBucket)) > 0
|
||||
}
|
||||
|
||||
// placements is every node's placement, assembled from its keys in either form (delivery.PlacementKeys).
|
||||
func (w *world) placements() map[string]delivery.Placement {
|
||||
out, _ := w.placement.AllRevisions()
|
||||
return out
|
||||
}
|
||||
|
||||
// placedNodes is every node the seat's placement names (ADR 0567 §1): those compose their own modules (ADR 0547 §2),
|
||||
@@ -333,17 +343,7 @@ type worldReader struct {
|
||||
}
|
||||
|
||||
func (r worldReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
||||
out := map[string]delivery.Placement{}
|
||||
for key, value := range r.w.read(delivery.PlacementBucket) {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if p, err := delivery.DecodePlacement(value); err == nil && p.Node == node {
|
||||
out[node] = p
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return r.w.placements(), nil
|
||||
}
|
||||
|
||||
func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030212912-6c5a8ef893b4
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030212912-6c5a8ef893b4 h1:uzXzKfjoWB8wCbAPgxUAWSArr+3DKwi8qrB3BUEOVOE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030212912-6c5a8ef893b4/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -96,8 +96,9 @@ func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Pla
|
||||
return p, out, findings
|
||||
}
|
||||
|
||||
// Write is one key a warrant writes (§5): the node, its key, the revision it is written at (0: it has none, so it is
|
||||
// created) and its value (nil: it is deleted).
|
||||
// Write is one node a warrant writes (§5): the node, its node key `<node>.node` (the commit of the node's keys,
|
||||
// mesh-sdk delivery.PlacementKeys.Writes), the revision of the node's placement it is written at (0: it has none, so
|
||||
// it is created) and its value (nil: it is deleted).
|
||||
type Write struct {
|
||||
Node, Key string
|
||||
Base uint64
|
||||
@@ -108,7 +109,7 @@ type Write struct {
|
||||
func Writes(p delivery.PlacementProposal, laid map[string]*delivery.Placement) ([]Write, error) {
|
||||
var out []Write
|
||||
for _, node := range p.Nodes() {
|
||||
key, err := delivery.PlacementKey(node)
|
||||
key, err := delivery.PlacementNodeKey(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -103,7 +103,7 @@ func TestThePlacementSeatDecides(t *testing.T) {
|
||||
}
|
||||
|
||||
writes, err := Writes(p, laid)
|
||||
if err != nil || len(writes) != 2 || writes[0].Key != "placement.anchor" || writes[0].Base != 4 || writes[1].Base != 9 {
|
||||
if err != nil || len(writes) != 2 || writes[0].Key != "anchor.node" || writes[0].Base != 4 || writes[1].Base != 9 {
|
||||
t.Fatalf("writes = %+v, %v", writes, err)
|
||||
}
|
||||
pinned := *laid["laptop"]
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-builder
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2 h1:RF73nVI1dTrrmlFosVDTMJmpUQ+rEeB7++BHH/RvZFg=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-checker
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2 h1:RF73nVI1dTrrmlFosVDTMJmpUQ+rEeB7++BHH/RvZFg=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -583,12 +583,13 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
|
||||
for _, m := range modules {
|
||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
value, err := delivery.EncodePlacement(p)
|
||||
// Its node key, the commit that names the modules placed (research 072 row 37).
|
||||
values, err := delivery.PlacementValues(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := placement.Put(ctx, key, value); err != nil {
|
||||
key, _ := delivery.PlacementNodeKey(node)
|
||||
if _, err := placement.Put(ctx, key, values[key]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2 h1:RF73nVI1dTrrmlFosVDTMJmpUQ+rEeB7++BHH/RvZFg=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-view
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2 h1:RF73nVI1dTrrmlFosVDTMJmpUQ+rEeB7++BHH/RvZFg=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -34,15 +34,13 @@ func place(t *testing.T, node string, modules ...string) Entry {
|
||||
for _, m := range modules {
|
||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
key, err := delivery.PlacementKey(node)
|
||||
// Its node key, which names the modules placed (research 072 row 37).
|
||||
values, err := delivery.PlacementValues(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v, err := delivery.EncodePlacement(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return Entry{Bucket: delivery.PlacementBucket, Key: key, Value: v, At: t0}
|
||||
key, _ := delivery.PlacementNodeKey(node)
|
||||
return Entry{Bucket: delivery.PlacementBucket, Key: key, Value: values[key], At: t0}
|
||||
}
|
||||
|
||||
func find(t *testing.T, s Snapshot, node, module string) Row {
|
||||
@@ -107,7 +105,7 @@ func TestLeftBehindIsWhatThePlacementDoesNotPut(t *testing.T) { // 1134/AC3, iss
|
||||
if find(t, s, "a", "w").Status != StatusLeftBehind || s.Counted() != 4 {
|
||||
t.Fatalf("a have with no placement: %+v counted %d", find(t, s, "a", "w"), s.Counted())
|
||||
}
|
||||
fill(m, Entry{Bucket: delivery.PlacementBucket, Key: "placement.b", Deleted: true})
|
||||
fill(m, Entry{Bucket: delivery.PlacementBucket, Key: "b.node", Deleted: true})
|
||||
if s := m.Snapshot(); s.Counted() != 2 || find(t, s, "b", "x").Status != StatusLeftBehind {
|
||||
t.Fatalf("b's placement deleted: counted %d, %+v", s.Counted(), s.Rows)
|
||||
}
|
||||
|
||||
@@ -4,8 +4,8 @@
|
||||
// list and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk buscompose).
|
||||
//
|
||||
// **Memberships are published for real** (the operator, 2026-10-25 22:31): since release 6 nobody publishes them, so a
|
||||
// module placed since cannot serve its seat or reach a new state. Each is published on the assignments stream only when
|
||||
// it differs from the one last published there. **A bucket, log or stream a manifest declares that the bus lacks is
|
||||
// module placed since cannot serve its seat or reach a new state. Each is published in the part's own state memberships,
|
||||
// and said on its subject, only when it differs from the one last published there. **A bucket, log or stream a manifest declares that the bus lacks is
|
||||
// created** (the operator, 2026-10-26 08:54; issue 1790), once per manifest, with the manifest's limits, on the event
|
||||
// that brought the manifest; one the bus holds is never changed. **The user list is written sealed to the bus node**
|
||||
// (ADR 0547 clause 7) as an object of the part's own bucket named in `user-list.sealed`, which the bus node follows only
|
||||
@@ -112,7 +112,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
bus := newBusReader(ctx, js)
|
||||
f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn},
|
||||
Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js},
|
||||
Publish: memberships{nc: nc}, Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js},
|
||||
Objects: sealedObjects{ctx: ctx, js: js}, Logf: logger.Printf,
|
||||
Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}}
|
||||
f.Watch = bus
|
||||
|
||||
@@ -1,55 +1,24 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
|
||||
// memberships publishes each membership on the assignments stream, last per subject, over the one last read there
|
||||
// (ExpectLastSequencePerSubject, so a membership another writer published meanwhile is not overwritten, issues 930 and
|
||||
// 934), and reads the last one by a direct get of its subject. The one place the part publishes on the bus.
|
||||
// memberships says each membership the part wrote to its state on its subject, `mesh.assignment.<node>.<module>`, as a
|
||||
// plain message: a notice for a runtime that follows the subject alone, which the assignments stream also keeps while it
|
||||
// exists. It reads nothing there and asks no stream for an acknowledgement, so the stream's removal at the lease write
|
||||
// (research 072 row 33) stops no membership (issue 774). The one place the part publishes a membership on the bus.
|
||||
type memberships struct {
|
||||
ctx context.Context
|
||||
nc *nats.Conn
|
||||
js jetstream.JetStream
|
||||
nc *nats.Conn
|
||||
}
|
||||
|
||||
func (m memberships) LastMembership(node, module string) ([]byte, uint64, bool, error) {
|
||||
ctx, cancel := context.WithTimeout(m.ctx, requestWithin)
|
||||
defer cancel()
|
||||
msg, err := m.nc.RequestWithContext(ctx, membership.DirectGetSubject(node, module), nil)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
err = fmt.Errorf("no answer in %s: a read the bus refuses is answered by none (%w)", requestWithin, err)
|
||||
}
|
||||
return nil, 0, false, err
|
||||
}
|
||||
switch status := msg.Header.Get("Status"); status {
|
||||
case "":
|
||||
case "404":
|
||||
return nil, 0, false, nil
|
||||
default:
|
||||
return nil, 0, false, fmt.Errorf("the assignments stream answered %s %s", status, msg.Header.Get("Description"))
|
||||
}
|
||||
seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
|
||||
if err != nil {
|
||||
return nil, 0, false, fmt.Errorf("the membership of %s on %s carries no sequence", module, node)
|
||||
}
|
||||
return msg.Data, seq, true, nil
|
||||
}
|
||||
|
||||
func (m memberships) PublishMembership(node, module string, body []byte, after uint64) error {
|
||||
ctx, cancel := context.WithTimeout(m.ctx, requestWithin)
|
||||
defer cancel()
|
||||
if _, err := m.js.Publish(ctx, membership.Subject(node, module), body, jetstream.WithExpectLastSequencePerSubject(after)); err != nil {
|
||||
return fmt.Errorf("the membership of %s on %s after sequence %d: %w", module, node, after, err)
|
||||
func (m memberships) PublishMembership(node, module string, body []byte) error {
|
||||
if err := m.nc.Publish(membership.Subject(node, module), body); err != nil {
|
||||
return fmt.Errorf("the membership of %s on %s: %w", module, node, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-grants
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031000557-748543c35f7f
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031000557-748543c35f7f h1:5I7+LiZPua0XrkiaLKWm4/3TyJUcwyJ3OA0p3WOoQos=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031000557-748543c35f7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -47,10 +47,8 @@ type fakeBus struct {
|
||||
calls []string
|
||||
events []string
|
||||
writes []string
|
||||
// The assignments stream: each subject's last membership and sequence, every publish, and whether it is refused.
|
||||
// Each membership's subject: the last membership said there, every notice, and whether it is refused.
|
||||
stream map[string][]byte
|
||||
seqs map[string]uint64
|
||||
seq uint64
|
||||
publishes []string
|
||||
refusePub bool
|
||||
// hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus.
|
||||
@@ -106,29 +104,17 @@ func (b *fakeBus) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) er
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
subject := "mesh.assignment." + node + "." + module
|
||||
v, ok := b.stream[subject]
|
||||
return v, b.seqs[subject], ok, nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) PublishMembership(node, module string, body []byte, after uint64) error {
|
||||
func (b *fakeBus) PublishMembership(node, module string, body []byte) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
subject := "mesh.assignment." + node + "." + module
|
||||
if b.refusePub {
|
||||
return errors.New("Permissions Violation for Publish to " + subject)
|
||||
}
|
||||
if b.seqs[subject] != after {
|
||||
return fmt.Errorf("wrong last sequence for %s: %d", subject, b.seqs[subject])
|
||||
}
|
||||
if b.stream == nil {
|
||||
b.stream, b.seqs = map[string][]byte{}, map[string]uint64{}
|
||||
b.stream = map[string][]byte{}
|
||||
}
|
||||
b.seq++
|
||||
b.stream[subject], b.seqs[subject] = body, b.seq
|
||||
b.stream[subject] = body
|
||||
b.publishes = append(b.publishes, subject)
|
||||
return nil
|
||||
}
|
||||
@@ -1082,6 +1068,15 @@ func TestGrants(t *testing.T) {
|
||||
if !slices.Contains(read.Assigned["b"], "mod-x") || read.Placed["b"].Node != "b" || slices.ContainsFunc(read.Gaps, func(g grants.Gap) bool { return g.Input == "seat holdings" }) {
|
||||
t.Errorf("read: assigned %v, placed %v, gaps %v", read.Assigned, read.Placed, read.Gaps)
|
||||
}
|
||||
// Research 072 row 37: a node key and a key per module, the legacy mirror left out, read the same.
|
||||
perModule, _ := delivery.PlacementValues(delivery.Placement{Node: "b", Modules: []delivery.PlacedModule{{Module: "mod-y"}, {Module: "mod-x"}}})
|
||||
delete(perModule, "placement.b")
|
||||
nb := fixture()
|
||||
nb.states[grants.ReadPlacement] = perModule
|
||||
if got := grants.ReadInputs(nb, &grants.Manifests{}); !slices.Equal(got.Placed["b"].Assigned(), []string{"mod-y", "mod-x"}) ||
|
||||
slices.ContainsFunc(got.Gaps, func(g grants.Gap) bool { return strings.HasPrefix(g.Input, grants.ReadPlacement) }) {
|
||||
t.Errorf("read of the keys per module: placed %v, gaps %v", got.Placed, got.Gaps)
|
||||
}
|
||||
b.refused[grants.ReadPlacement] = true
|
||||
if read := grants.ReadInputs(b, &grants.Manifests{}); slices.ContainsFunc(read.Gaps, func(g grants.Gap) bool { return g.Input == grants.ReadPlacement && g.Failed }) {
|
||||
t.Errorf("placement not on the bus is a failed read: %v", read.Gaps)
|
||||
@@ -1764,10 +1759,11 @@ func TestGrants(t *testing.T) {
|
||||
b.manifestOf("mod-x", strings.Replace(xManifest, `"own-secrets"`, `"claims":`+row.claims+`,"own-secrets"`, 1))
|
||||
}
|
||||
b.loaded = shadowOf(t, b)
|
||||
// The last membership is the part's own state's, never the assignments stream's: the stream goes at the lease
|
||||
// write (research 072 row 33), and a part reading its last there would publish nothing after it.
|
||||
if row.last != nil {
|
||||
raw, _ := json.Marshal(row.last(composed(b)))
|
||||
b.stream = map[string][]byte{"mesh.assignment." + key: raw}
|
||||
b.seqs = map[string]uint64{"mesh.assignment." + key: 4}
|
||||
b.own[grants.StateMemberships] = map[string][]byte{key: raw}
|
||||
}
|
||||
res := partOn(b).Pass()
|
||||
if got := slices.Contains(res.Published, key); got != row.published {
|
||||
@@ -1775,12 +1771,12 @@ func TestGrants(t *testing.T) {
|
||||
}
|
||||
// The state holds the bare membership the assignments stream carries, through mesh-sdk's codec, and
|
||||
// only one published (issue 774); what became of it is the kept pass's, which grants_membership reads.
|
||||
state, _, inState, _ := b.ownGet(grants.StateMemberships, key)
|
||||
state, _, _, _ := b.ownGet(grants.StateMemberships, key)
|
||||
ms, _ := partOn(b).MembershipOf("a", "mod-x")
|
||||
pass := fmt.Sprint(ms.(map[string]any)["last_pass"])
|
||||
if row.published {
|
||||
if row.names == "" && row.last != nil && !strings.Contains(string(b.stream["mesh.assignment."+key]), "10.0.0.1") {
|
||||
t.Errorf("the private network of the last membership is not carried: %s", b.stream["mesh.assignment."+key])
|
||||
if row.names == "" && row.last != nil && !strings.Contains(string(state), "10.0.0.1") {
|
||||
t.Errorf("the private network of the last membership is not carried: %s", state)
|
||||
}
|
||||
got, err := membership.DecodeGrantsMembership(state)
|
||||
var sent membership.Membership
|
||||
@@ -1793,7 +1789,7 @@ func TestGrants(t *testing.T) {
|
||||
}
|
||||
return
|
||||
}
|
||||
if inState {
|
||||
if was, _ := json.Marshal(row.last(composed(b))); !bytes.Equal(state, was) {
|
||||
t.Errorf("a membership held back reached the state: %s", state)
|
||||
}
|
||||
said := strings.Join(res.NotPublished, "\n")
|
||||
@@ -1805,18 +1801,21 @@ func TestGrants(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("a refused publish: no exit, said, published at the next event", func(t *testing.T) {
|
||||
// The assignments stream removed (nats_stream_remove at the lease write) and its subject refused: every membership
|
||||
// is still written to the state, where each runtime reads it, and the refusal is said (issue 774).
|
||||
t.Run("the stream gone and its subject refused: memberships still written, the refusal said", func(t *testing.T) {
|
||||
b := fixture()
|
||||
b.loaded = shadowOf(t, b)
|
||||
b.refusePub = true
|
||||
p := partOn(b)
|
||||
refused := p.Compose()
|
||||
if len(refused.Published) != 0 || len(refused.NotPublished) != 3 || !strings.Contains(refused.NotPublished[0], "Permissions Violation") {
|
||||
t.Errorf("a refused publish = %+v: said, and nothing published", refused)
|
||||
if !slices.Equal(refused.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) || refused.MembershipsWritten != 3 ||
|
||||
!strings.Contains(strings.Join(refused.Errors, "\n"), "Permissions Violation") {
|
||||
t.Errorf("with the stream gone = %+v: every membership written, the refusal said", refused)
|
||||
}
|
||||
b.refusePub = false
|
||||
if next := p.Compose(); !slices.Equal(next.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) {
|
||||
t.Errorf("the next event published %v", next.Published)
|
||||
if next := p.Compose(); len(next.Published) != 0 || next.MembershipsWritten != 0 {
|
||||
t.Errorf("the next event published %v again", next.Published)
|
||||
}
|
||||
})
|
||||
t.Run("a tool answers while a pass waits on the bus", func(t *testing.T) {
|
||||
@@ -1965,16 +1964,15 @@ func TestGrants(t *testing.T) {
|
||||
b := fixture()
|
||||
b.loaded = shadowOf(t, b)
|
||||
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
||||
b.stream, b.seqs = map[string][]byte{}, map[string]uint64{}
|
||||
for i, key := range []string{"a.mod-x", "b.mod-y"} {
|
||||
b.own[grants.StateMemberships] = map[string][]byte{}
|
||||
for _, key := range []string{"a.mod-x", "b.mod-y"} {
|
||||
raw, _ := json.Marshal(grants.Memberships(recs)[key].Membership)
|
||||
var m map[string]any
|
||||
_ = json.Unmarshal(raw, &m)
|
||||
seat, verb := brokerSeat()
|
||||
m["seats"] = []any{map[string]any{"seat": seat, "verb": verb, "subject": "mesh.seat." + seat + ".tool." + verb}}
|
||||
raw, _ = json.Marshal(m)
|
||||
subject := "mesh.assignment." + key
|
||||
b.stream[subject], b.seqs[subject] = raw, uint64(i+1)
|
||||
b.own[grants.StateMemberships][key] = raw
|
||||
}
|
||||
var logged []string
|
||||
var lmu sync.Mutex
|
||||
|
||||
@@ -270,17 +270,25 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
|
||||
if keys, err := r.Keys(ReadPlacement); err != nil {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement, Why: "not read yet: the assignments alone place modules (" + err.Error() + ")"})
|
||||
} else {
|
||||
// Each node's keys in either form, assembled by the seat's one reader (delivery.PlacementsOf; research 072
|
||||
// row 37): its node key and a key per module, or its legacy key `placement.<node>`.
|
||||
values := map[string][]byte{}
|
||||
for _, k := range keys {
|
||||
raw, found, err := r.Get(ReadPlacement, k)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
p, err := delivery.DecodePlacement(raw)
|
||||
node, kerr := delivery.ParsePlacementKey(k)
|
||||
if err != nil || kerr != nil || p.Node != node {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement + " " + k, Why: "not a placement, so the node is composed from its assignments alone"})
|
||||
if _, _, _, err := delivery.ParsePlacementKey(k); err != nil {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement + " " + k, Why: "not a placement key, so it places nothing"})
|
||||
continue
|
||||
}
|
||||
values[k] = raw
|
||||
}
|
||||
placed, err := delivery.PlacementsOf(values)
|
||||
if err != nil {
|
||||
in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement, Why: "a key does not read, so its node is composed from what does: " + err.Error()})
|
||||
}
|
||||
for node, p := range placed {
|
||||
if in.Placed == nil {
|
||||
in.Placed = map[string]delivery.Placement{}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"slices"
|
||||
"sort"
|
||||
@@ -25,12 +24,12 @@ var MembershipGaps = []Gap{
|
||||
{Input: "membership: root-free machines", Why: "asked of the controller's root-free verb at each pass and kept for that pass alone; a machine it does not call free (an error included) is composed as not free, and the pass says why"},
|
||||
}
|
||||
|
||||
// Publisher publishes a membership on the assignments stream (membership.Subject), last per subject, and reads the one
|
||||
// last published there. A refusal (the grant not yet on the bus) is an error the part says and tries again at the next
|
||||
// event.
|
||||
// Publisher says a membership the part wrote to its state memberships on its subject (membership.Subject), as a plain
|
||||
// message, for a runtime that follows the subject alone. The state is where a membership is published and read back
|
||||
// (ADR 0571 clause 7); the subject is a notice that depends on no stream, so the assignments stream's removal at the
|
||||
// lease write (research 072 row 33) stops nothing. A refusal is said and the membership stays written.
|
||||
type Publisher interface {
|
||||
LastMembership(node, module string) (body []byte, seq uint64, found bool, err error)
|
||||
PublishMembership(node, module string, body []byte, after uint64) error
|
||||
PublishMembership(node, module string, body []byte) error
|
||||
}
|
||||
|
||||
// Membership is one module's membership on one node, as composed now, and what became of its publishing.
|
||||
@@ -73,18 +72,20 @@ const (
|
||||
MembershipUnchanged = "unchanged"
|
||||
)
|
||||
|
||||
// publish publishes m when it differs from the membership last published for its assignment, carrying that one's private
|
||||
// network and received contributions; it answers m as published (or why not) and whether it was published now. held is
|
||||
// every holding the memberships composed now carry (Holdings.ByMemberships).
|
||||
func publish(p Publisher, m Membership, h Holdings, held map[string]bool) (Membership, bool) {
|
||||
// publish says whether m is to be published: whether it differs from was, the membership last published (the value of
|
||||
// its key in the part's own state memberships, found when it is there), carrying that one's private network and
|
||||
// received contributions. It answers m as composed and published, or why not. held is every holding the memberships
|
||||
// composed now carry (Holdings.ByMemberships). Pure: the caller writes and notifies.
|
||||
func publish(p Publisher, was []byte, found bool, m Membership, h Holdings, held map[string]bool) (Membership, bool) {
|
||||
if p == nil {
|
||||
m.Published = "not published: no publisher here (only the part's process publishes)"
|
||||
return m, false
|
||||
}
|
||||
was, seq, found, err := p.LastMembership(m.Node, m.Module)
|
||||
if err != nil {
|
||||
m.Published = "not published: the membership last published cannot be read: " + err.Error()
|
||||
return m, false
|
||||
if found {
|
||||
// A value of the state is the bare membership, or the wrapper written before 2026-10-30 (mesh-sdk's one codec);
|
||||
// one that is neither is no last membership, and m is a first one.
|
||||
bare, err := membership.BareGrantsMembership(was)
|
||||
was, found = bare, err == nil
|
||||
}
|
||||
if found {
|
||||
var last membership.Membership
|
||||
@@ -122,10 +123,6 @@ func publish(p Publisher, m Membership, h Holdings, held map[string]bool) (Membe
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := p.PublishMembership(m.Node, m.Module, body, seq); err != nil {
|
||||
m.Published = fmt.Sprintf("not published (tried again at the next event): %v", err)
|
||||
return m, false
|
||||
}
|
||||
m.Published = MembershipPublished
|
||||
return m, true
|
||||
}
|
||||
|
||||
@@ -140,7 +140,7 @@ type Result struct {
|
||||
}
|
||||
|
||||
// NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
|
||||
const NothingTakesEffect = "memberships (on the assignments stream, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), and each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54); the user list, the grants and any change to a stream the bus holds stay in shadow until take-over"
|
||||
const NothingTakesEffect = "memberships (in its own state memberships, each said on its subject, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), and each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54); the user list, the grants and any change to a stream the bus holds stay in shadow until take-over"
|
||||
|
||||
// Part is the grants part: it reads, calls and writes through these, and nothing else.
|
||||
type Part struct {
|
||||
@@ -328,7 +328,33 @@ func (p *Part) Compose() Result {
|
||||
}
|
||||
seated := holdings.ByMemberships(members...)
|
||||
for _, key := range sortedKeys(ms) {
|
||||
m, now := publish(p.Publish, ms[key], holdings, seated)
|
||||
// The last membership is the state's own (ADR 0571 clause 7), never the assignments stream's, which the bus
|
||||
// module's tools remove at the lease write (research 072 row 33): a last read there would freeze every membership
|
||||
// once it is gone.
|
||||
was, _, found, err := p.Store.Get(StateMemberships, key)
|
||||
if err != nil {
|
||||
res.NotPublished = append(res.NotPublished, key+": not published: the membership last published cannot be read from its own "+StateMemberships+": "+err.Error())
|
||||
continue
|
||||
}
|
||||
m, now := publish(p.Publish, was, found, ms[key], holdings, seated)
|
||||
// Published is written to the state, where every runtime reads it (process.FollowMembership); only one
|
||||
// published, so a membership held back never reaches a reader. The subject is a notice after it.
|
||||
if now {
|
||||
body := membership.EncodeGrantsMembership(m.Membership)
|
||||
if _, err := p.Store.Put(StateMemberships, key, body); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be written: %v", StateMemberships, key, err))
|
||||
m.Published, now = "not published (tried again at the next event): its own "+StateMemberships+" cannot be written", false
|
||||
} else {
|
||||
res.MembershipsWritten++
|
||||
if err := p.Publish.PublishMembership(m.Node, m.Module, body); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("the membership of %s is written, its notice on its subject not said: %v", key, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
// One unchanged is written bare once where the state still holds the wrapper written before 2026-10-30.
|
||||
if m.Published == MembershipUnchanged && p.changed(&res, StateMemberships, key, membership.EncodeGrantsMembership(m.Membership)) {
|
||||
res.MembershipsWritten++
|
||||
}
|
||||
switch {
|
||||
case now:
|
||||
res.Published = append(res.Published, key)
|
||||
@@ -342,14 +368,6 @@ func (p *Part) Compose() Result {
|
||||
}
|
||||
res.Retired[key] = m.Retired
|
||||
}
|
||||
// The state holds the bare membership as the assignments stream carries it, through mesh-sdk's one codec (issue
|
||||
// 774): only one published now or unchanged, so a membership held back never reaches a reader by this way. What
|
||||
// became of each at this pass is the pass's own (Published, NotPublished, Retired), kept with it (Pass).
|
||||
if now || m.Published == MembershipUnchanged {
|
||||
if p.changed(&res, StateMemberships, key, membership.EncodeGrantsMembership(m.Membership)) {
|
||||
res.MembershipsWritten++
|
||||
}
|
||||
}
|
||||
}
|
||||
held, heldErr := HeldStreams(p.Call)
|
||||
var ask Ask
|
||||
|
||||
Reference in New Issue
Block a user