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

Placed: a module in the first revision heard of a node's key stands since before it, so the seed's placement does not leave out what already ran (issue 774)

Browse Source
This commit is contained in:
jschoubben
2026-10-28 21:54:04 +01:00
parent 1386219d4d
commit 18a47ac3c7
3 changed files with 24 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/delivery/buildrun/standing.go
-1
View File
@@ -94,7 +94,6 @@ func (g *ToolchainGate) Placement(key string, value []byte, deleted bool, at tim
func (g *ToolchainGate) Settled() bool {
g.mu.Lock()
defer g.mu.Unlock()
g.placed.Settle()
g.known = true
return g.all()
}
go/delivery/buildrun/standing_test.go
+12 -9
View File
@@ -126,8 +126,10 @@ func TestToolchainGate(t *testing.T) {
} {
t.Run(c.name, func(t *testing.T) {
gate, set, said := start(t, true)
k, v := placing(t, "ace") // ace's key, first heard placing nothing
gate.Placement(k, v, false, old.Add(-time.Hour))
gate.Settled()
k, v := placing(t, "ace", NodeBuildModule)
k, v = placing(t, "ace", NodeBuildModule)
if gate.Placement(k, v, false, placed) {
t.Fatal("a placement with no report passed something on")
}
@@ -173,23 +175,24 @@ func TestToolchainGate(t *testing.T) {
}
})
t.Run("the history: a full history's oldest revision stands since before it", func(t *testing.T) {
t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) {
p := delivery.NewPlaced()
k, v := placing(t, "ace", NodeBuildModule)
for i := range delivery.PlacementHistory {
p.Put(k, v, false, old.Add(time.Duration(i)*time.Minute))
}
k2, v2 := placing(t, "g14", NodeBuildModule)
p.Put(k, v, false, old)
k2, v2 := placing(t, "g14")
p.Put(k2, v2, false, old)
k2, v2 = placing(t, "g14", NodeBuildModule)
if _, changed, _ := p.Put(k2, v2, false, placed); strings.Join(changed, ",") != NodeBuildModule {
t.Errorf("a module placed says changed %v", changed)
}
if _, _, err := p.Put("placement.novox", []byte("not json"), false, old); err == nil {
t.Error("a value that does not read was not said")
}
p.Settle()
if s, ok := p.Since("ace", NodeBuildModule); !ok || !s.IsZero() {
t.Errorf("ace's module stands since %v, %v; want the zero time", s, ok)
}
if s, ok := p.Since("g14", NodeBuildModule); !ok || !s.Equal(old) {
t.Errorf("g14's module stands since %v, %v; want its first revision", s, ok)
if s, ok := p.Since("g14", NodeBuildModule); !ok || !s.Equal(placed) {
t.Errorf("g14's module stands since %v, %v; want the revision that placed it", s, ok)
}
if got := p.PlacedNodes(NodeBuildModule); strings.Join(got, ",") != "ace,g14" || len(p.ByNode()) != 2 || !p.Seeded() {
t.Errorf("PlacedNodes = %v, ByNode = %v", got, p.ByNode())
go/delivery/placed.go
+12 -36
View File
@@ -14,23 +14,22 @@ import (
// with the controller's lease, ADR 0549 §6).
//
// Its input is the state's revisions in the order the bus wrote them, as a watch with the history (jetstream's
// IncludeHistory) hands them: Put takes each, and Settle ends the history. A module stands since the revision of its
// node's key that first named it in an unbroken run; a module in the oldest revision of a key whose history is full
// (PlacementHistory) stands since before what the bus still keeps (the zero time), since an earlier revision may have
// named it. A key's value that does not read places nothing on its node, and Put says why.
// IncludeHistory) hands them: Put takes each. A module stands since the revision of its node's key that first named it
// in an unbroken run. A module in the first revision heard of a key stands since the zero time: that revision is the
// key's first (the seed of ADR 0549 §4 placed what already ran there) or the oldest the bus still keeps
// (PlacementHistory), and either way the module may have run there before it. A key's value that does not read places
// nothing on its node, and Put says why.
//
// Safe for use from a watch and from a reader at once.
type Placed struct {
mu sync.Mutex
since map[string]map[string]time.Time // node -> module -> since when it stands
seen map[string]int // node -> revisions read before Settle
first map[string]time.Time // node -> when its first revision read was written
settled bool
mu sync.Mutex
since map[string]map[string]time.Time // node -> module -> since when it stands
heard map[string]bool // the nodes whose key was heard at all
}
// NewPlaced is a Placed that holds no placement yet.
func NewPlaced() *Placed {
return &Placed{since: map[string]map[string]time.Time{}, seen: map[string]int{}, first: map[string]time.Time{}}
return &Placed{since: map[string]map[string]time.Time{}, heard: map[string]bool{}}
}
// Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote
@@ -57,11 +56,9 @@ func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node
}
p.mu.Lock()
defer p.mu.Unlock()
if !p.settled {
if p.seen[node] == 0 {
p.first[node] = at
}
p.seen[node]++
if !p.heard[node] {
p.heard[node] = true
at = time.Time{}
}
was := p.since[node]
next := map[string]time.Time{}
@@ -87,27 +84,6 @@ func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node
return node, changed, err
}
// Settle ends the history: from now on every Put is a new revision. A module in the oldest kept revision of a key whose
// history is full stands since the zero time.
func (p *Placed) Settle() {
p.mu.Lock()
defer p.mu.Unlock()
if p.settled {
return
}
p.settled = true
for node, n := range p.seen {
if n < PlacementHistory {
continue
}
for m, s := range p.since[node] {
if s.Equal(p.first[node]) {
p.since[node][m] = time.Time{}
}
}
}
}
// Seeded says some node's placement is held: before the first placement file reaches the indexer nothing is placed,
// and a reader keeps what it did before (ADR 0549 §5).
func (p *Placed) Seeded() bool {