diff --git a/modules/disk-load/cmd/disk-load/data.go b/modules/disk-load/cmd/disk-load/data.go index 724cec2d..7882c804 100644 --- a/modules/disk-load/cmd/disk-load/data.go +++ b/modules/disk-load/cmd/disk-load/data.go @@ -115,6 +115,8 @@ func (w *Watched) Follow(k *Keeper) error { return nil }) w.Ready() + // The handed-over set is complete now: a module whose key went while the watch was down is retired once. + k.RetireUnkeyed(context.Background()) // A round refused while the watch caught up runs now, not at the next hour (the delivery's readings within minutes). go k.RoundIfRefused(context.Background()) return nil diff --git a/modules/disk-load/cmd/disk-load/data_test.go b/modules/disk-load/cmd/disk-load/data_test.go index 5d375056..c8a86f46 100644 --- a/modules/disk-load/cmd/disk-load/data_test.go +++ b/modules/disk-load/cmd/disk-load/data_test.go @@ -134,6 +134,37 @@ func TestTheItemsComeFromTheEnginesKeysAndARoundSaysHowFarItGot(t *testing.T) { if measured["/srv/web2"] != before || round().Items != 2 { t.Errorf("an item whose key was deleted was measured again: %d times, round %+v", measured["/srv/web2"]-before, round()) } + + // A restart: oldie's key was deleted while the watch was down, so no delete arrives. Once the watch handed over its + // keys, oldie's reading is retired and pg's kept; an empty set retires nothing (the engine has not written yet). + _ = readings.Put("ace.oldie.x", health.Reading{V: health.ReadingVersion, Node: "ace", + PlacedDataItem: catalogue.PlacedDataItem{Module: "oldie", ID: "x", Path: "/srv/oldie"}}) + for _, restart := range []struct { + name string + keys []stdio.StateChange + retired bool + }{ + {"no key yet", nil, false}, + {"pg's key alone", []stdio.StateChange{{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}}, true}, + } { + w2 := &Watched{Node: "ace"} + said = nil + k2 := &Keeper{Node: "ace", Items: w2.All, Store: readings, Say: k.Say, Measurer: k.Measurer} + for _, ch := range restart.keys { + if _, _, err := w2.Take(ch); err != nil { + t.Fatal(err) + } + } + w2.Ready() + k2.RetireUnkeyed(context.Background()) + oldie, pgData := readingOf(readings, "ace.oldie.x"), readingOf(readings, "ace.pg.data") + if oldie == nil || pgData == nil || oldie.RetiredAt.IsZero() == restart.retired || !pgData.RetiredAt.IsZero() { + t.Errorf("%s: oldie %+v, pg %+v; oldie retired: want %v, pg never", restart.name, oldie, pgData, restart.retired) + } + if !restart.retired && !strings.Contains(strings.Join(said, "\n"), "holds no key") { + t.Errorf("%s: the skip was not said: %v", restart.name, said) + } + } } func readingOf(s *memStore, key string) *health.Reading { diff --git a/modules/disk-load/cmd/disk-load/keeper.go b/modules/disk-load/cmd/disk-load/keeper.go index 5d9e7a16..95998f3f 100644 --- a/modules/disk-load/cmd/disk-load/keeper.go +++ b/modules/disk-load/cmd/disk-load/keeper.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" "os" "os/exec" @@ -286,13 +287,47 @@ func (k *Keeper) MeasureModule(ctx context.Context, module string) { // asked only on a change of that module's key (novox/hq ADR 0577 clause 2), never from a round. It is cleared when the // item is declared again (measureOne). func (k *Keeper) retireGone(module string, declared []catalogue.PlacedDataItem) (int, error) { + return k.retire(func(r health.Reading) bool { return r.Module == module }, declared) +} + +// RetireUnkeyed retires, once the watch handed over every current key, the readings of this node whose module has no +// key in that set: a key deleted while the watch was not up (disk-load restarted, the machine rebooted or slept) never +// reaches disk-load as a delete. An empty set is skipped and said: the engine has not written its keys yet, and an +// empty set would retire every reading. +func (k *Keeper) RetireUnkeyed(ctx context.Context) { + k.mu.Lock() + defer k.mu.Unlock() + items, err := k.Items(ctx) + if err != nil { + k.Say("the readings of modules with no key are not looked for: %v", err) + return + } + if len(items) == 0 { + k.Say("the node-engine's state %s holds no key of this machine yet: no reading is retired for it", delivery.EngineDataBucket) + return + } + var declared []catalogue.PlacedDataItem + for _, its := range items { + declared = append(declared, its...) + } + n, err := k.retire(func(r health.Reading) bool { _, keyed := items[r.Module]; return !keyed }, declared) + switch { + case err != nil: + k.Say("the readings of modules with no key are not all retired: %v", err) + case n > 0: + k.Say("%d reading(s) of modules with no key of this machine retired, measured no more", n) + } +} + +// retire marks retired every reading of this node that of picks and declared does not name. +func (k *Keeper) retire(of func(health.Reading) bool, declared []catalogue.PlacedDataItem) (int, error) { keys, err := k.Store.Keys() if err != nil { return 0, err } declaredNow := map[string]bool{} for _, it := range declared { - declaredNow[health.ReadingKey(k.Node, module, it.ID)] = true + declaredNow[health.ReadingKey(k.Node, it.Module, it.ID)] = true } now := k.Measurer.Now().UTC() n := 0 @@ -305,7 +340,7 @@ func (k *Keeper) retireGone(module string, declared []catalogue.PlacedDataItem) continue } r, err := health.DecodeReading(raw) - if err != nil || r.Node != k.Node || r.Module != module || !r.RetiredAt.IsZero() { + if err != nil || r.Node != k.Node || !of(r) || !r.RetiredAt.IsZero() { continue } r.RetiredAt = now