151 lines
4.5 KiB
Go
151 lines
4.5 KiB
Go
package delivery
|
|
|
|
import (
|
|
"fmt"
|
|
"maps"
|
|
"slices"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Placed is where the indexer's placement (the state `placement`, ADR 0549 §2, §3) places each module, and since when:
|
|
// the one reader of that state for every part that asks "does module stand on node" (issue 774, design 47's third
|
|
// milestone), in place of each part reading the controller's assignments (`mesh-controller_assignments`, which goes
|
|
// 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.
|
|
//
|
|
// 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
|
|
}
|
|
|
|
// 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{}}
|
|
}
|
|
|
|
// Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote
|
|
// it). It answers the node the key is of and the modules whose standing on it changed (placed or removed), in name
|
|
// order, and an error for a key or a value that does not read (the node then places nothing).
|
|
func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node string, changed []string, err error) {
|
|
node, err = ParsePlacementKey(key)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
now := map[string]bool{}
|
|
if !deleted {
|
|
pl, derr := DecodePlacement(value)
|
|
switch {
|
|
case derr != nil:
|
|
err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr)
|
|
case pl.Node != node:
|
|
err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, pl.Node)
|
|
default:
|
|
for _, m := range pl.Assigned() {
|
|
now[m] = true
|
|
}
|
|
}
|
|
}
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
if !p.settled {
|
|
if p.seen[node] == 0 {
|
|
p.first[node] = at
|
|
}
|
|
p.seen[node]++
|
|
}
|
|
was := p.since[node]
|
|
next := map[string]time.Time{}
|
|
for m := range now {
|
|
if s, ok := was[m]; ok {
|
|
next[m] = s
|
|
} else {
|
|
next[m] = at
|
|
changed = append(changed, m)
|
|
}
|
|
}
|
|
for m := range was {
|
|
if !now[m] {
|
|
changed = append(changed, m)
|
|
}
|
|
}
|
|
if len(next) == 0 {
|
|
delete(p.since, node)
|
|
} else {
|
|
p.since[node] = next
|
|
}
|
|
slices.Sort(changed)
|
|
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 {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return len(p.since) > 0
|
|
}
|
|
|
|
// Since says whether module stands on node, and since when (the zero time: since before the revisions kept).
|
|
func (p *Placed) Since(node, module string) (time.Time, bool) {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
s, ok := p.since[node][module]
|
|
return s, ok
|
|
}
|
|
|
|
// PlacedNodes is every node module is placed on, in name order.
|
|
func (p *Placed) PlacedNodes(module string) []string {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
var out []string
|
|
for node, mods := range p.since {
|
|
if _, ok := mods[module]; ok {
|
|
out = append(out, node)
|
|
}
|
|
}
|
|
slices.Sort(out)
|
|
return out
|
|
}
|
|
|
|
// ByNode is, by node, the modules placed on it in name order.
|
|
func (p *Placed) ByNode() map[string][]string {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
out := make(map[string][]string, len(p.since))
|
|
for node, mods := range p.since {
|
|
out[node] = slices.Sorted(maps.Keys(mods))
|
|
}
|
|
return out
|
|
}
|