mesh-self-check: D13 judges the data from disk-load's readings, restic's last good nights and the indexer's manifests, with the controller's kinds, keys and words (issue 1952, ADR 0233)

This commit is contained in:
2026-10-29 23:54:54 +01:00
parent 89f63b3ab8
commit f7f02abbb3
4 changed files with 524 additions and 7 deletions
@@ -34,7 +34,7 @@ const (
// `reads`, which its test holds to these.
var (
Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb}
Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead, HealthRead}
Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead, HealthRead, ReadingsRead, GoodRead, ModulesRead, ManifestsRead}
)
// Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe
@@ -65,7 +65,9 @@ func Registry() []Probe {
Deferred: "the bindings of provisions to their consumers: " + storeOnly},
{ID: "D13", Asserts: "every item of data a machine declares is measured, is there, holds what it held, and is " +
"backed up within its bound", From: "issue 273, ADR 0233", Kind: "data-shrank",
Deferred: "the data items' measurements and their history: " + storeOnly},
Raises: []string{kindDataMissing, kindDataQuiet, kindBackupStale, kindArrayDegraded, kindProtectionMissing,
kindDataUnmeasured},
Run: probeData},
{ID: "D15", Asserts: "no user of the bus had its connection dropped more than twelve times in the last hour: the " +
"bus module's nats_closed_connections", From: "issue 327", Kind: "bus-reconnects", Run: probeReconnects},
{ID: "DA", Asserts: "every machine that names an agent account has it judged, on its node-engine's newest " +
@@ -76,9 +78,7 @@ func Registry() []Probe {
Deferred: "the node-engines' statements of root and the tools that run as root: " + storeOnly},
{ID: "D-fill", Asserts: "every module's bucket and log holds less than three quarters of its cap; one said " +
"filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501",
Kind: "bucket-or-log-filling", Raises: []string{"bucket-or-log-unread"},
Deferred: "each bucket's and log's cap: the bus module's nats_buckets answers bytes and no cap, and the " +
"declared caps are " + storeOnly},
Kind: "bucket-or-log-filling", Run: probeFill},
{ID: "D-heard", Asserts: "no grant of a restricted stream names a module that composes nothing for it",
From: "ADR 0376 §2", Kind: "grant-composes-nothing",
Deferred: "the grants of restricted streams with what each machine composed for them: " + storeOnly},
@@ -0,0 +1,342 @@
package main
// D13, the data the modules declare (novox/hq ADR 0233, ADR 0571 clause 3, ADR 0564 §6): every item of data a machine
// declares is measured, is there, holds what it held, and is backed up within its bound. The controller composed what
// every machine declares and asked each machine's backup holder what it measured; the facts now live on the bus: the
// readings disk-load keeps (its state `readings`, with the item's seven-day peak), the last good night per item restic
// keeps (its state `good`), and the manifests the indexer holds (for the bounds an item declares). Each kind, key and
// word is the controller's (mesh-controller cmd/mesh-controller/data.go), so the condition the keeper raises from it is
// the one the controller raised.
import (
"context"
"encoding/json"
"fmt"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
// The states D13 reads (declared under `reads`).
const (
ReadingsRead = health.ReadingsRead
GoodRead = health.GoodRead
ModulesRead = delivery.IndexerModule + ".modules"
ManifestsRead = delivery.IndexerModule + ".manifests"
)
// The kinds D13 raises.
const (
kindDataShrank = "data-shrank"
kindEmptyReplacement = "empty-replacement"
kindDataHeldTwice = "data-held-twice"
kindDataQuiet = "data-quiet"
kindBackupStale = "backup-stale"
kindDataUnmeasured = "data-unmeasured"
kindDataMissing = "data-missing"
kindArrayDegraded = "array-degraded"
kindProtectionMissing = "protection-missing"
kindCleanupWaiting = "cleanup-waiting"
shrinkWindowDays = 7
shrinkFloor int64 = 16 << 20
)
// dataFacts is everything D13 judges: the readings, the last good nights, each module's manifest, and the machines whose
// backup holder was heard from.
type dataFacts struct {
readings []health.Reading
good map[string]time.Time // by health.GoodKey
manifests map[string]catalogue.Manifest
// unmeasured are the machines that hold node-backup, are heard from, and have no reading of anything.
unmeasured []string
now time.Time
}
// dataFacts reads them from the bus.
func (c *Checker) dataFacts() (dataFacts, error) {
f := dataFacts{good: map[string]time.Time{}, manifests: map[string]catalogue.Manifest{}, now: c.Now()}
keys, err := c.Store.Keys(ReadingsRead)
if err != nil {
return f, fmt.Errorf("disk-load's readings cannot be read: %w", err)
}
byNode := map[string]int{}
for _, k := range keys {
raw, ok, err := c.Store.Get(ReadingsRead, k)
if err != nil {
return f, fmt.Errorf("the reading %s cannot be read: %w", k, err)
}
if !ok {
continue
}
r, err := health.DecodeReading(raw)
if err != nil {
continue
}
f.readings = append(f.readings, r)
byNode[r.Node]++
}
gkeys, err := c.Store.Keys(GoodRead)
if err != nil {
return f, fmt.Errorf("restic's last good nights cannot be read: %w", err)
}
for _, k := range gkeys {
raw, ok, err := c.Store.Get(GoodRead, k)
if err != nil {
return f, fmt.Errorf("the last good night %s cannot be read: %w", k, err)
}
if g, derr := health.DecodeGoodRow(raw); ok && derr == nil && !g.LastGood.IsZero() {
f.good[health.GoodKey(g.Node, g.Module, g.Item)] = g.LastGood
}
}
mods, err := c.Store.Keys(ModulesRead)
if err != nil {
return f, fmt.Errorf("the indexer's modules cannot be read: %w", err)
}
for _, name := range mods {
raw, ok, err := c.Store.Get(ModulesRead, name)
if err != nil || !ok {
continue
}
// Only the manifest's digest of the indexer's module, and the manifest's text: what the bounds are read from.
var m struct {
Manifest string `json:"manifest"`
}
if json.Unmarshal(raw, &m) != nil || m.Manifest == "" {
continue
}
mraw, ok, err := c.Store.Get(ManifestsRead, m.Manifest)
if err != nil || !ok {
continue
}
var doc struct {
JSON string `json:"manifest"`
}
if json.Unmarshal(mraw, &doc) != nil {
continue
}
var man catalogue.Manifest
if json.Unmarshal([]byte(doc.JSON), &man) == nil {
f.manifests[name] = man
}
}
heard, err := c.heard()
if err != nil {
return f, err
}
holders, err := c.holdersOf("node-backup")
if err != nil {
return f, err
}
for _, n := range holders {
if heard[n] && byNode[n] == 0 {
f.unmeasured = append(f.unmeasured, n)
}
}
return f, nil
}
// probeData is D13.
func probeData(ctx context.Context, c *Checker) ([]Observation, error) {
f, err := c.dataFacts()
if err != nil {
return nil, err
}
return dataFindings(f), nil
}
func severityOf(class string) health.Severity {
if class == catalogue.ClassIrreplaceable {
return Urgent
}
return Warning
}
func keyOfItem(machine, module, item string) string { return machine + "." + module + "." + item }
func comparable(precision string) bool {
return precision == "" || precision == "exact" || strings.HasPrefix(precision, "dataset ")
}
func datasetOf(precision string) string {
rest, ok := strings.CutPrefix(precision, "dataset ")
if !ok {
return ""
}
name, _, _ := strings.Cut(rest, ":")
return name
}
func orUnknownPath(p string) string {
if p == "" {
return "a path its backup holder never named"
}
return p
}
func sizeWords(size *int64) string {
if size == nil || *size < 0 {
return "size unknown"
}
b := float64(*size)
for _, unit := range []string{"B", "KB", "MB", "GB"} {
if b < 1024 || unit == "GB" {
if unit == "B" {
return fmt.Sprintf("%d B", *size)
}
return fmt.Sprintf("%.1f %s", b, unit)
}
b /= 1024
}
return ""
}
// firstSeen is when an item was first read: the oldest day its seven-day peak is made of, else its measurement. A lower
// bound on how long it has been declared (the controller kept the day it first saw it).
func firstSeen(r health.Reading, now time.Time) time.Time {
if len(r.Days) > 0 {
if t, err := time.Parse("2006-01-02", r.Days[0].Day); err == nil {
return t
}
}
if r.MeasuredAt != nil {
return *r.MeasuredAt
}
return now
}
func obs(kind string, id, machine string, sev health.Severity, summary, said string) Observation {
return Observation{Kind: kind, Token: kind, Severity: sev, Summary: summary, Said: said,
Subject: Subject{Scope: ScopeMachine, ID: id, Machine: machine}}
}
// dataFindings is every condition the data on record raises now: a function of what is known, as the controller's was.
func dataFindings(f dataFacts) []Observation {
var out []Observation
arrays := map[string][]health.Reading{}
type shrunk struct {
machine, dataset, class string
size, peak int64
items []string
}
shrunkDatasets := map[string]shrunk{}
for _, m := range f.unmeasured {
o := obs(kindDataUnmeasured, m, m, Warning, fmt.Sprintf("%s's backup holder did not say what it measured of the data "+
"declared there, so nothing about that data is known", m), "disk-load keeps no reading of this machine's data")
o.Confirm = true
out = append(out, o)
}
rs := append([]health.Reading(nil), f.readings...)
sort.Slice(rs, func(i, j int) bool {
return keyOfItem(rs[i].Node, rs[i].Module, rs[i].ID) < keyOfItem(rs[j].Node, rs[j].Module, rs[j].ID)
})
for _, r := range rs {
class := r.Class
item, declared := f.manifests[r.Module].DataItem(r.ID)
id := keyOfItem(r.Node, r.Module, r.ID)
if !catalogue.Watched(class) {
continue
}
severity := severityOf(class)
if r.Redundancy != nil {
where := r.Node + "/" + r.Redundancy.Kind + ":" + r.Redundancy.Where
arrays[where] = append(arrays[where], r)
} else if declared && item.Redundancy != "" && r.MeasuredAt != nil && r.Error == "" {
out = append(out, obs(kindProtectionMissing, id, r.Node, severity,
fmt.Sprintf("%s of %s on %s (%s) is said to be protected by the redundancy of the storage it is on, "+
"and it is on nothing the backup holder can read as redundant: it has no protection the mesh can see",
r.ID, r.Module, r.Node, class),
orUnknownPath(r.Path)+" is on no redundant storage the backup holder can read"))
}
peak := r.Peak
shrank := peak != nil && r.SizeBytes != nil && comparable(r.Precision) && *r.SizeBytes*2 < *peak && *peak-*r.SizeBytes >= shrinkFloor
switch {
case r.Error != "" && strings.Contains(r.Error, "does not exist"):
out = append(out, obs(kindDataMissing, id, r.Node, severity,
fmt.Sprintf("%s of %s on %s (%s) is gone: where it was declared, nothing exists any more", r.ID, r.Module, r.Node, class),
orUnknownPath(r.Path)+" does not exist: "+firstLine(r.Error)))
case shrank && datasetOf(r.Precision) != "":
// Several items on one dataset share its size: one condition for the dataset, as loud as the most
// precious item on it.
k := r.Node + "/" + datasetOf(r.Precision)
ds := shrunkDatasets[k]
ds.machine, ds.dataset, ds.size, ds.peak = r.Node, datasetOf(r.Precision), *r.SizeBytes, *peak
ds.class = catalogue.StricterClass(ds.class, class)
ds.items = append(ds.items, r.Module+"/"+r.ID)
shrunkDatasets[k] = ds
case shrank:
out = append(out, obs(kindDataShrank, id, r.Node, severity,
fmt.Sprintf("%s of %s on %s (%s) shrank to %s from %s within %d days — more than half of what it "+
"held is gone. If that was meant, silence this with why; if not, `node-backup.restore` puts the last "+
"good copy beside it", r.ID, r.Module, r.Node, class, sizeWords(r.SizeBytes), sizeWords(peak), shrinkWindowDays), ""))
}
if !declared {
continue
}
if within := item.ActiveWithin(); within > 0 && r.LastWrite != nil && f.now.Sub(*r.LastWrite) > within {
out = append(out, obs(kindDataQuiet, id, r.Node, severity,
fmt.Sprintf("%s of %s on %s is written all the time, and has not been since %s (its bound is %s): "+
"whatever writes it has stopped", r.ID, r.Module, r.Node, r.LastWrite.UTC().Format(time.RFC3339), within), ""))
}
// A backup is required of what is irreplaceable and copied; of what is valuable it is the standard
// plan, said only where the machine was measured — where a holder is there to take it.
if item.BackedUp() && (class == catalogue.ClassIrreplaceable || r.MeasuredAt != nil) {
within := item.BackupWithin()
last, has := f.good[health.GoodKey(r.Node, r.Module, r.ID)]
switch {
case !has && f.now.Sub(firstSeen(r, f.now)) > within:
out = append(out, obs(kindBackupStale, id, r.Node, severity,
fmt.Sprintf("%s of %s on %s is %s and has no good backup on record, %s after it was first "+
"declared — is node-backup held there, and do its nights succeed? (`node-backup.backed-up`)",
r.ID, r.Module, r.Node, class, f.now.Sub(firstSeen(r, f.now)).Round(time.Hour)), ""))
case has && f.now.Sub(last) > within:
out = append(out, obs(kindBackupStale, id, r.Node, severity,
fmt.Sprintf("%s of %s on %s is %s and its newest good backup is from %s, older than its bound "+
"of %s", r.ID, r.Module, r.Node, class, last.UTC().Format(time.RFC3339), within), ""))
}
}
}
keys := make([]string, 0, len(shrunkDatasets))
for k := range shrunkDatasets {
keys = append(keys, k)
}
sort.Strings(keys)
for _, k := range keys {
ds := shrunkDatasets[k]
out = append(out, obs(kindDataShrank, ds.machine+".dataset."+strings.ReplaceAll(ds.dataset, "/", "-"), ds.machine,
severityOf(ds.class), fmt.Sprintf("the dataset on %s that holds %s shrank to %s from %s within %d days — more than "+
"half of what it held is gone", ds.machine, strings.Join(ds.items, ", "), sizeWords(&ds.size), sizeWords(&ds.peak),
shrinkWindowDays), "the dataset "+ds.dataset))
}
// The redundant storage watched data is on: one condition per array, as loud as the most precious item on it —
// the array, not each item, is what degrades.
wheres := make([]string, 0, len(arrays))
for w := range arrays {
wheres = append(wheres, w)
}
sort.Strings(wheres)
for _, where := range wheres {
items := arrays[where]
red := items[0].Redundancy
if red.Healthy != nil && *red.Healthy {
continue
}
class, machine := catalogue.ClassValuable, items[0].Node
var names []string
for _, r := range items {
class = catalogue.StricterClass(class, r.Class)
names = append(names, r.Module+"/"+r.ID)
}
state := "could not be read"
if red.Healthy != nil {
state = "is NOT healthy"
}
out = append(out, obs(kindArrayDegraded, machine+".array."+strings.NewReplacer("/", "-", ":", "-").Replace(red.Kind+"-"+red.Where),
machine, severityOf(class), fmt.Sprintf("the %s storage on %s that protects %s %s", red.Kind, machine,
strings.Join(names, ", "), state), fmt.Sprintf("the %s storage %s %s: %s", red.Kind, red.Where, state, firstLine(red.Said))))
}
return out
}
@@ -7,9 +7,12 @@ import (
"slices"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
// bus is the bus module's answers as a probe asks them: every mesh stream and consumer as defined, unless changed.
@@ -21,6 +24,8 @@ type bus struct {
asksFail error
// banned is each node's answer to node-intrusion-prevention.banned; a node absent does not answer.
banned map[string]string
// fills are rows nats_streams answers beside the mesh's streams: a module's bucket or log with its size and cap.
fills []fillRow
}
func (b bus) rows() map[string]*streamRow {
@@ -73,6 +78,16 @@ func (b bus) ask(key string, body any) (json.RawMessage, error) {
detail = append(detail, *r)
}
}
if stream == "" && len(b.fills) > 0 {
var rows []any
for _, r := range detail {
rows = append(rows, r)
}
for _, f := range b.fills {
rows = append(rows, f)
}
return json.Marshal(map[string]any{"detail": rows})
}
if stream != "" && len(detail) == 0 {
return nil, errors.New("no stream called " + stream)
}
@@ -152,6 +167,40 @@ func TestTheBusProbes(t *testing.T) {
"ace": `{"banned":[]}`}}, nil, "", machines()},
{"D8 a heard holder that does not answer is no pass", probeBans, bus{banned: map[string]string{}}, nil,
"a ban list could not be read", machines()},
// D13 (issue 273, ADR 0233): the readings disk-load keeps and the last good nights restic keeps, against the bounds
// the manifests declare; the kinds, keys and words are the controller's.
{"D13 all held", probeData, bus{}, nil, "", dataMesh(func(*dataRig) {})},
{"D13 shrank, gone, quiet, unprotected and stale", probeData, bus{}, []string{
"machine.ace.pg.data.data-quiet", "machine.ace.pg.data.backup-stale", "machine.ace.pg.dumps.data-missing",
"machine.ace.plex.movies.protection-missing", "machine.ace.plex.thumbs.data-shrank",
"machine.ace.array.md-md127.array-degraded"},
"", dataMesh(func(r *dataRig) {
r.size("ace", "pg", "data", 100<<20, 100<<20).write("ace", "pg", "data", -80*time.Hour)
r.gone("ace", "pg", "dumps")
r.size("ace", "plex", "thumbs", 10<<20, 600<<20)
r.size("ace", "plex", "movies", 1<<30, 1<<30).noArray("ace", "plex", "movies")
r.size("ace", "plex", "library", 1<<30, 1<<30).array("ace", "plex", "library", "md", "md127", false)
r.lastGood("ace", "pg", "data", -100*time.Hour)
})},
{"D13 a dataset shrank once for its items", probeData, bus{}, []string{"machine.ace.dataset.storage-media.data-shrank"},
"", dataMesh(func(r *dataRig) {
r.size("ace", "plex", "thumbs", 10<<20, 600<<20).dataset("ace", "plex", "thumbs", "storage/media")
r.size("ace", "plex", "meta", 10<<20, 600<<20).dataset("ace", "plex", "meta", "storage/media")
})},
{"D13 a machine with a backup holder and no reading", probeData, bus{}, []string{"machine.g14.data-unmeasured"}, "",
dataMesh(func(r *dataRig) { r.holder("g14") })},
// D-fill (ADR 0297 §6): a module's bucket or log at three quarters of its cap, kept until under seven tenths.
{"D-fill all with room", probeFill, bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 74, MaxBytes: 100},
{Name: "LOG_mesh-delivery-builder_build-log", Bytes: 10, MaxBytes: 100}, {Name: "KV_nocap_x", Bytes: 999}}}, nil, "", nil},
{"D-fill a bucket and a log filling", probeFill, bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 75, MaxBytes: 100},
{Name: "LOG_mesh-delivery-builder_build-log", Bytes: 99, MaxBytes: 100}}},
[]string{"bus.KV_mesh-issues_issues.filling", "bus.LOG_mesh-delivery-builder_build-log.filling"}, "", nil},
{"D-fill between the thresholds raises nothing new", probeFill,
bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 72, MaxBytes: 100}}}, nil, "", nil},
{"D-fill between the thresholds keeps what is said", probeFill,
bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 72, MaxBytes: 100}}},
[]string{"bus.KV_mesh-issues_issues.filling"}, "", filling("KV_mesh-issues_issues")},
{"D-fill the bus module silent", probeFill, bus{asksFail: errors.New("no responders")}, nil, "did not answer", nil},
// DA (ADR 0266): the agent account each machine names, judged on its engine's report while it is heard from.
{"DA confined", probeAgentAccounts, bus{}, nil, "", agents(map[string]string{"novox": stateHealthy})},
{"DA a way to root found", probeAgentAccounts, bus{}, []string{"machine.novox.agent-root"}, "",
@@ -166,7 +215,7 @@ func TestTheBusProbes(t *testing.T) {
agents(map[string]string{"novox": "old"})},
}
for _, tc := range cases {
c := &Checker{Ask: tc.bus.ask, Store: tc.mesh}
c := &Checker{Ask: tc.bus.ask, Store: tc.mesh, Now: func() time.Time { return dataNow }}
if tc.mesh == nil {
c.Store = newStore()
}
@@ -205,6 +254,16 @@ func machines() *memStore {
return s
}
// filling is a store holding the self-check's written fill finding of a stream: the condition open.
func filling(stream string) *memStore {
s := newStore()
o := Observation{Kind: "bucket-or-log-filling", Token: "filling", Subject: Subject{Scope: ScopeBus, ID: stream},
Summary: "filling", Source: "D-fill"}
raw, _ := health.EncodeFinding(o.Finding("run", time.Now()))
_ = s.Put(FindingsState, o.Key(), json.RawMessage(raw))
return s
}
// agents is machines() where each node named names the agent account "agent" and its report holds the verdict on it:
// a state, "" for none, "pending" for one waiting for its search, "old" for an engine older than the judging.
func agents(verdicts map[string]string) *memStore {
@@ -255,3 +314,115 @@ func TestTheRegistryIsADR0564sTable(t *testing.T) {
t.Errorf("the registry holds %v, ADR 0564 moves %v", ids, moved)
}
}
// ---- D13's rig ------------------------------------------------------------------------------------------------
var dataNow = time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
// dataRig is the mesh D13 reads: a manifest per module with the items it declares, a reading per item, a last good night.
type dataRig struct {
s *memStore
readings map[string]*health.Reading
good map[string]time.Time
}
func (r *dataRig) key(node, module, item string) string { return health.ReadingKey(node, module, item) }
func (r *dataRig) reading(node, module, item string) *health.Reading {
k := r.key(node, module, item)
if r.readings[k] == nil {
at := dataNow
r.readings[k] = &health.Reading{V: health.ReadingVersion, Node: node, PlacedDataItem: catalogue.PlacedDataItem{Module: module,
ID: item, Class: catalogue.ClassValuable, Path: "/srv/" + module + "/" + item}, Measured: health.Measured{MeasuredAt: &at, Precision: "exact"}}
}
return r.readings[k]
}
func (r *dataRig) size(node, module, item string, size, peak int64) *dataRig {
x := r.reading(node, module, item)
x.SizeBytes, x.Peak = &size, &peak
return r
}
func (r *dataRig) write(node, module, item string, ago time.Duration) *dataRig {
t := dataNow.Add(ago)
r.reading(node, module, item).LastWrite = &t
return r
}
func (r *dataRig) gone(node, module, item string) *dataRig {
r.reading(node, module, item).Error = "/srv/" + module + "/" + item + " does not exist"
return r
}
func (r *dataRig) dataset(node, module, item, name string) *dataRig {
r.reading(node, module, item).Precision = "dataset " + name + ": its whole size"
return r
}
func (r *dataRig) array(node, module, item, kind, where string, healthy bool) *dataRig {
r.reading(node, module, item).Redundancy = &health.Redundancy{Kind: kind, Where: where, Healthy: &healthy, Said: "a member is missing"}
return r
}
func (r *dataRig) noArray(node, module, item string) *dataRig {
r.reading(node, module, item).Redundancy = nil
return r
}
func (r *dataRig) lastGood(node, module, item string, ago time.Duration) *dataRig {
r.good[r.key(node, module, item)] = dataNow.Add(ago)
return r
}
func (r *dataRig) holder(node string) {
key, _ := delivery.SeatHoldersKey("node-backup", node)
_ = r.s.Put(SeatHoldersRead, key, delivery.SeatHolders{V: 1, Seat: "node-backup", Scope: "node", Node: node,
Holders: []delivery.SeatHolder{{Module: "restic", Node: node}}})
_ = r.s.Put(PresenceRead, node, delivery.Presence{V: 1, Online: true})
}
// dataMesh builds the mesh: pg declares data (backed up within 48 h, written all the time within 24 h) and dumps; plex
// declares movies (protected by redundancy), thumbs, meta and library (all valuable); every item is read as held, and what
// is given changes it.
func dataMesh(change func(*dataRig)) *memStore {
r := &dataRig{s: newStore(), readings: map[string]*health.Reading{}, good: map[string]time.Time{}}
r.holder("ace")
declare := func(module string, items map[string]catalogue.DataItem) {
var man catalogue.Manifest
man.Module = module
man.Data = &catalogue.Data{}
for id, it := range items {
it.ID = id
man.Data.Own = append(man.Data.Own, it)
}
raw, _ := json.Marshal(man)
digest := delivery.ManifestDigest(raw)
_ = r.s.Put(ManifestsRead, digest, delivery.Manifest{V: 1, Module: module, JSON: string(raw)})
_ = r.s.Put(ModulesRead, module, delivery.Module{V: 1, Manifest: digest})
}
declare("pg", map[string]catalogue.DataItem{
"data": {Class: catalogue.ClassValuable, Within: "48h", Active: "24h"},
"dumps": {Class: catalogue.ClassValuable}})
declare("plex", map[string]catalogue.DataItem{
"movies": {Class: catalogue.ClassValuable, Redundancy: "the array"},
"thumbs": {Class: catalogue.ClassValuable},
"meta": {Class: catalogue.ClassValuable},
"library": {Class: catalogue.ClassValuable, Redundancy: "the array"}})
for _, i := range [][2]string{{"pg", "data"}, {"pg", "dumps"}, {"plex", "movies"}, {"plex", "thumbs"}, {"plex", "meta"}, {"plex", "library"}} {
r.size("ace", i[0], i[1], 1<<30, 1<<30).lastGood("ace", i[0], i[1], -time.Hour).write("ace", i[0], i[1], -time.Minute)
if i[0] == "plex" && (i[1] == "movies" || i[1] == "library") {
r.array("ace", i[0], i[1], "md", "md127", true)
}
}
change(r)
for k, x := range r.readings {
_ = r.s.Put(ReadingsRead, k, x)
}
for k, t := range r.good {
node, rest, _ := strings.Cut(k, ".")
module, item, _ := strings.Cut(rest, ".")
_ = r.s.Put(GoodRead, k, health.GoodRow{V: health.GoodVersion, Node: node, Module: module, Item: item, CoveredBy: "/x", LastGood: t})
}
return r.s
}
+5 -1
View File
@@ -22,7 +22,11 @@
"mesh-delivery-assigner.placement",
"mesh-delivery-assigner.seat-holders",
"mesh-delivery-watchman.presence",
"mesh-host.health"
"mesh-host.health",
"disk-load.readings",
"restic.good",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.manifests"
],
"state": [
"findings",