jschoubben
4f51393b72
ADR 0593 §6: the provisioner, the assigner and the view read compositions instead of composing, so the node's engine writes each one it makes to its composition state: the wanted builds, every placed module at its newest done build (so the plan exists before the want moves), each with its provisioning plan, and a module refused alone with its refusal. A key is written only when it differs, and kept while among the module's five newest or named by a want. A placed module's trunk build landing now makes a pass. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
104 lines
2.9 KiB
Go
104 lines
2.9 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-host/internal/selfcompose"
|
|
)
|
|
|
|
// ---- this node's composition state, in the bus's own storage (novox/hq ADR 0593 §6) --------------------------------
|
|
|
|
// composedStore binds the node's composition state `composed-<node>` over the link open now: bound, never created (the
|
|
// grants part creates it, create-only, history delivery.ComposedHistory).
|
|
func composedStore(js nats.JetStreamContext, node string) (nats.KeyValue, error) {
|
|
bucket, err := delivery.ComposedBucketFor(node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
kv, err := js.KeyValue(bucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("this node's composition state %s cannot be bound (the grants part creates it; written "+
|
|
"again at the next pass that composes): %w", bucket, err)
|
|
}
|
|
return kv, nil
|
|
}
|
|
|
|
func (l *natsLink) composedStore(node string) (nats.KeyValue, error) {
|
|
return composedStore(l.js, node)
|
|
}
|
|
|
|
func (b OverNATS) composedStore(node string) (nats.KeyValue, error) { return composedStore(b.JS, node) }
|
|
|
|
// EngineComposed is this node's composition state over whichever link the session has open (selfcompose.ComposedStore);
|
|
// with none open every read and write is refused, and nothing is kept anywhere else.
|
|
type EngineComposed struct {
|
|
Session *Session
|
|
Node string
|
|
}
|
|
|
|
var _ selfcompose.ComposedStore = EngineComposed{}
|
|
|
|
func (e EngineComposed) kv() (nats.KeyValue, error) {
|
|
e.Session.init()
|
|
e.Session.mu.Lock()
|
|
bus := e.Session.bus
|
|
e.Session.mu.Unlock()
|
|
l, ok := bus.(interface {
|
|
composedStore(node string) (nats.KeyValue, error)
|
|
})
|
|
if bus == nil || !ok {
|
|
return nil, ErrNoLink
|
|
}
|
|
return l.composedStore(e.Node)
|
|
}
|
|
|
|
// Entries is every key the state holds with its latest value and revision.
|
|
func (e EngineComposed) Entries(ctx context.Context) (map[string]selfcompose.ComposedEntry, error) {
|
|
kv, err := e.kv()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
keys, err := kv.Keys(nats.Context(ctx))
|
|
if errors.Is(err, nats.ErrNoKeysFound) {
|
|
return map[string]selfcompose.ComposedEntry{}, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make(map[string]selfcompose.ComposedEntry, len(keys))
|
|
for _, key := range keys {
|
|
got, err := kv.Get(key)
|
|
if errors.Is(err, nats.ErrKeyNotFound) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out[key] = selfcompose.ComposedEntry{Value: got.Value(), Revision: got.Revision()}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Put writes a key and answers its revision.
|
|
func (e EngineComposed) Put(_ context.Context, key string, value []byte) (uint64, error) {
|
|
kv, err := e.kv()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return kv.Put(key, value)
|
|
}
|
|
|
|
// Delete deletes a key (a delete marker, kept with the key's history).
|
|
func (e EngineComposed) Delete(_ context.Context, key string) error {
|
|
kv, err := e.kv()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return kv.Delete(key)
|
|
}
|