mesh-catalog/modules/disk-load/cmd/disk-load/data.go

package main

// Where disk-load learns the data items it measures (novox/hq ADR 0577 clause 2): the node-engine's state data
// (engineDataBucket), one key `<node>.<module>` per module placed on this node that declares data items, as the
// engine composed them. disk-load watches its own node's keys and composes nothing: it reads no resolved values, no
// manifests and no placement, so a manifest its mesh-sdk cannot judge never silences it, and a round reads one watched
// map instead of several hundred keys over the tunnel (issue 1955).

import (
	"context"
	"errors"
	"fmt"
	"sync"

	engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
	nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
)

// The node-engine's state data as the bus names it: the seat's local name under its holder (the one name of the holder
// here, held equal to the manifest's `reads` by a unit row).
const (
	engineHolder     = "mesh-host"
	engineDataBucket = engineHolder + "_" + nodeengine.StateData
	engineDataRead   = engineHolder + "." + nodeengine.StateData
)

// itemsOf are the seat's data items as this module measures them.
func itemsOf(in []nodeengine.PlacedDataItem) []catalogue.PlacedDataItem {
	out := make([]catalogue.PlacedDataItem, 0, len(in))
	for _, it := range in {
		out = append(out, catalogue.PlacedDataItem{Module: it.Module, ID: it.Item, Class: it.Class, Path: it.Path,
			CoveredBy: it.CoveredBy, Protection: it.Protection, Measure: it.Measure})
	}
	return out
}

// Watched is this node's data items as the engine's state data holds them, by module, kept from a watch of its keys.
type Watched struct {
	Node string

	mu       sync.Mutex
	byKey    map[string]string // key to its module, for a delete that carries no value
	byModule map[string][]catalogue.PlacedDataItem
	ready    bool
}

// ErrNotWatchedYet is a round asked before the watch handed over the engine's current keys: no item is known yet, and
// none is taken for gone.
var ErrNotWatchedYet = errors.New("the node-engine's state " + engineDataBucket + " is not watched yet: " +
	"no data item of this machine is known")

// Take keeps one change of a key and answers the module it is of and whether its items changed.
func (w *Watched) Take(ch stdio.StateChange) (string, bool, error) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.byKey == nil {
		w.byKey, w.byModule = map[string]string{}, map[string][]catalogue.PlacedDataItem{}
	}
	if ch.Op != stdio.OpPut {
		module, known := w.byKey[ch.Key]
		if !known {
			return "", false, nil
		}
		delete(w.byKey, ch.Key)
		delete(w.byModule, module)
		return module, true, nil
	}
	d, err := engineclient.DecodeData(ch.Value)
	if err != nil {
		return "", false, fmt.Errorf("%s: %w", ch.Key, err)
	}
	if d.Node != w.Node {
		return "", false, fmt.Errorf("%s names the machine %q, and this is %s", ch.Key, d.Node, w.Node)
	}
	items := itemsOf(d.Items)
	for _, it := range items {
		if it.Module != d.Module {
			return "", false, fmt.Errorf("%s holds an item of %q, and is %s's", ch.Key, it.Module, d.Module)
		}
	}
	changed := !sameItems(w.byModule[d.Module], items)
	w.byKey[ch.Key] = d.Module
	w.byModule[d.Module] = items
	return d.Module, changed, nil
}

// Ready says the watch handed over every current key: from now on an item missing is an item gone.
func (w *Watched) Ready() {
	w.mu.Lock()
	w.ready = true
	w.mu.Unlock()
}

// All is every module's items, as the Keeper reads them; refused until the watch is ready.
func (w *Watched) All(context.Context) (map[string][]catalogue.PlacedDataItem, error) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if !w.ready {
		return nil, ErrNotWatchedYet
	}
	out := make(map[string][]catalogue.PlacedDataItem, len(w.byModule))
	for m, its := range w.byModule {
		out[m] = append([]catalogue.PlacedDataItem(nil), its...)
	}
	return out, nil
}

func sameItems(a, b []catalogue.PlacedDataItem) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// Follow watches this node's keys of the engine's state data, and asks the keeper to measure a module whose items changed
// after the watch began; it returns once the current keys were handed over (stdio.KeptState.Follow).
func (w *Watched) Follow(k *Keeper) error {
	pattern, err := engineclient.DataNodeKeys(w.Node)
	if err != nil {
		return err
	}
	stdio.State(engineDataRead).Follow(pattern, func(ch stdio.StateChange) error {
		module, changed, err := w.Take(ch)
		if err != nil || !changed || ch.Current {
			return err
		}
		k.Due(module)
		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
}