mesh-host/internal/selfcompose/data.go

package selfcompose

import (
	"bytes"
	"context"
	"fmt"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/compose"
	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// **The data items of each module placed here, as this engine composed them** (novox/hq ADR 0577 clause 1). After a
// pass that composed and handed something new (or the first one of a run), the composer works out each placed module's
// data items from the same inputs it composed from (frombus.DataItems, which reads them through this composer's own
// view of the bus) and keeps each module's in the engine's state data, key `<node>.<module>`, with the digest of the
// module's composition: written only when the value differs, deleted when the module no longer declares any here or left
// the node. That node's disk-load watches its node's keys and measures what they name; it composes nothing (clause 2).

// DataStore is this node's keys of the engine's state data (link.EngineData).
type DataStore interface {
	// Get is a key's value, nil when it has none.
	Get(ctx context.Context, key string) ([]byte, error)
	Put(ctx context.Context, key string, value []byte) error
	Delete(ctx context.Context, key string) error
	// Keys is every key matching pattern that holds a value.
	Keys(ctx context.Context, pattern string) ([]string, error)
}

// itemsReader is the composer's view of the bus for working out data items: no own secret is asked of the vault, since
// working out where an item lies reads no secret (the items are rendered as a plan is, compose.Planning).
type itemsReader struct{ reader }

func (itemsReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
	return nil, nil
}

// dataItemsOf works out the data items of the scope's node and, by module, why each module held for a manifest the
// indexer refused is left out (frombus.DataItemsHeld, ADR 0578 §3); a test sets what it answers.
var dataItemsOf = frombus.DataItemsHeld

// publishData keeps the data items of this node's modules in the engine's state data, when a pass asked it (dataDue).
func (c *Composer) publishData(ctx context.Context) {
	c.mu.Lock()
	due := c.dataDue
	c.dataDue = false
	c.mu.Unlock()
	if !due || c.env.Data == nil {
		return
	}
	c.mu.Lock()
	scope := c.dataScope
	c.mu.Unlock()
	if scope.Node == "" {
		scope.Node = c.env.Node
	}
	items, unplaced, held, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope)
	if err != nil {
		c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " +
			delivery.EngineDataBucket + " holds stands; " + c.oweData() + ": " + err.Error())
		return
	}
	// A module held for a refused manifest declares nothing this pass can read: it is named, its other modules' items
	// are kept as ever, and its own key stands as it was until the manifest is accepted (ADR 0578 §3).
	if len(held) > 0 {
		var lines []string
		for _, m := range heldModules(held) {
			lines = append(lines, m+": "+held[m])
		}
		c.env.Say("data items of " + strings.Join(heldModules(held), ", ") + " not worked out, as the manifest is refused " +
			"and the module held (its key stands as it was): " + strings.Join(lines, "; "))
	}
	if len(unplaced) > 0 {
		c.env.Say("data items left out as not placeable: " + strings.Join(unplaced, "; "))
	}
	c.mu.Lock()
	digests := c.digestsByModule()
	c.mu.Unlock()
	wrote, removed, err := c.writeData(ctx, items, digests, held)
	if err != nil {
		c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket +
			"; " + c.oweData() + ": " + err.Error())
		return
	}
	c.mu.Lock()
	c.dataTries = 0
	c.mu.Unlock()
	c.sayData("")
	if wrote+removed > 0 {
		c.env.Say(fmt.Sprintf("the data items of this node's modules: %d key(s) of %s written, %d deleted",
			wrote, delivery.EngineDataBucket, removed))
	}
}

// oweData keeps the data items due and has the composer's loop make a pass again after the growing bound (LaterWait), as
// a pass refused for a bus that did not answer is: the first write after a delivery may be refused while the state or
// its grant is not on the bus yet, and nothing else would try again until an unrelated composition changed. It answers
// the words that say when.
func (c *Composer) oweData() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.dataDue = true
	c.wait = LaterWait(c.dataTries)
	c.dataTries++
	c.owed.Store(true)
	return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.dataTries)
}

// sayData says a line about the data items once until it changes; "" clears it.
func (c *Composer) sayData(line string) {
	c.mu.Lock()
	same := c.dataSaid == line
	c.dataSaid = line
	c.mu.Unlock()
	if !same && line != "" {
		c.env.Say(line)
	}
}

// digestsByModule is the digest of each module's composition handed last, and the node's own under "" for a module run
// inside it. Under c.mu.
func (c *Composer) digestsByModule() map[string]string {
	out := map[string]string{}
	for key, dig := range c.handed {
		_, module, _, isNode, err := delivery.ParseComposedKey(key)
		if err != nil {
			continue
		}
		if isNode {
			module = ""
		}
		out[module] = dig
	}
	return out
}

// writeData writes each module's key whose value differs from what the key holds, and deletes this node's keys of
// modules with no items here now. The keys are read once per run (dataKept), so a restart that finds the same items
// writes nothing.
func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.PlacedDataItem, digests map[string]string, held map[string]string) (wrote, removed int, err error) {
	store := c.env.Data
	if c.dataKept == nil {
		pattern, err := delivery.EngineDataNodeKeys(c.env.Node)
		if err != nil {
			return 0, 0, err
		}
		keys, err := store.Keys(ctx, pattern)
		if err != nil {
			return 0, 0, fmt.Errorf("this node's keys cannot be listed: %w", err)
		}
		kept := map[string][]byte{}
		for _, k := range keys {
			raw, err := store.Get(ctx, k)
			if err != nil {
				return 0, 0, fmt.Errorf("%s cannot be read: %w", k, err)
			}
			if raw != nil {
				kept[k] = raw
			}
		}
		c.dataKept = kept
	}
	want := map[string][]byte{}
	modules := make([]string, 0, len(items))
	for m := range items {
		modules = append(modules, m)
	}
	sort.Strings(modules)
	var failed []string
	for _, module := range modules {
		if len(items[module]) == 0 {
			continue
		}
		key, err := delivery.EngineDataKey(c.env.Node, module)
		if err != nil {
			failed = append(failed, err.Error())
			continue
		}
		digest, ok := digests[module]
		if !ok {
			digest = digests[""] // run inside the node's own composition
		}
		raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: c.env.Node, Module: module, Digest: digest, Items: items[module]})
		if err != nil {
			failed = append(failed, err.Error())
			continue
		}
		want[key] = raw
	}
	keys := make([]string, 0, len(want))
	for k := range want {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, key := range keys {
		if bytes.Equal(c.dataKept[key], want[key]) {
			continue
		}
		if err := store.Put(ctx, key, want[key]); err != nil {
			failed = append(failed, key+": "+err.Error())
			continue
		}
		c.dataKept[key] = want[key]
		wrote++
	}
	var gone []string
	for key := range c.dataKept {
		if _, still := want[key]; !still && !heldKey(c.env.Node, key, held) {
			gone = append(gone, key)
		}
	}
	sort.Strings(gone)
	for _, key := range gone {
		if err := store.Delete(ctx, key); err != nil {
			failed = append(failed, key+": "+err.Error())
			continue
		}
		delete(c.dataKept, key)
		removed++
	}
	if len(failed) > 0 {
		return wrote, removed, fmt.Errorf("%s", strings.Join(failed, "; "))
	}
	return wrote, removed, nil
}

// heldKey is whether key is the key of a module held for a refused manifest, which stands as it was.
func heldKey(node, key string, held map[string]string) bool {
	for module := range held {
		if k, err := delivery.EngineDataKey(node, module); err == nil && k == key {
			return true
		}
	}
	return false
}