package selfcompose import ( "bytes" "context" "fmt" "sort" "strings" "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 `.`, 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 (frombus.DataItems); a test sets what it answers. var dataItemsOf = frombus.DataItems // 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, 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: " + err.Error()) return } 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) if err != nil { c.mu.Lock() c.dataDue = true // the next pass tries again, whatever it hands c.mu.Unlock() c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket + ", and are written again at the next composition: " + err.Error()) return } 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)) } } // 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) (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 { 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 }