Judge a have with no composition against the module's slice of the declaration the node last applied: equal is unchanged (issue 1945; ADR 0540 clause 9)
The first want at the switch skips the record hold only where it changes nothing on the node. A slice differing only in the bus's user list is marked user-list-only, so no yes waits on it at the switch; any other change at the same fingerprint is judged as any move, against the declaration's digest.
This commit is contained in:
@@ -2,18 +2,23 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// composeModules composes each named node and writes its compositions that differ (ADR 0539 clause 5): what each act
|
||||
@@ -156,6 +161,7 @@ func openComposedPlace(ctx context.Context, js jetstream.JetStream, conn *nats.C
|
||||
ran: func(ctx context.Context, module string) (sdkdelivery.Have, bool, error) {
|
||||
return haveOn(ctx, js, node, module)
|
||||
},
|
||||
lastApplied: lastAppliedOn(js, node),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -278,3 +284,79 @@ type composedPlan struct {
|
||||
catalogue.Plan
|
||||
RequiredBy string `json:"required-by,omitempty"`
|
||||
}
|
||||
|
||||
// lastAppliedOn reads, once, the declaration the node was last sent from the kept stream of declarations (the last
|
||||
// message on `mesh.node.<node>.declare`), and answers each module's slice of it (its resources that say the module, or
|
||||
// whose id names it) with the declaration's digest. None when the stream or the node's message is gone.
|
||||
func lastAppliedOn(js jetstream.JetStream, node string) func(ctx context.Context, module string) ([]json.RawMessage, string, bool, error) {
|
||||
var once sync.Once
|
||||
var last []byte
|
||||
var found bool
|
||||
var readErr error
|
||||
return func(ctx context.Context, module string) ([]json.RawMessage, string, bool, error) {
|
||||
once.Do(func() { last, found, readErr = lastDeclaration(ctx, js, node) })
|
||||
if readErr != nil || !found {
|
||||
return nil, "", false, readErr
|
||||
}
|
||||
return sliceOfDeclaration(last, module)
|
||||
}
|
||||
}
|
||||
|
||||
// lastDeclaration is the node's last signed declaration on the kept stream, as the stream holds it.
|
||||
func lastDeclaration(ctx context.Context, js jetstream.JetStream, node string) ([]byte, bool, error) {
|
||||
stream, err := js.Stream(ctx, broker.DeclarationsStream)
|
||||
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return nil, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
msg, err := stream.GetLastMsgForSubject(ctx, "mesh.node."+node+".declare")
|
||||
if errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
return nil, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return msg.Data, true, nil
|
||||
}
|
||||
|
||||
// sliceOfDeclaration is module's slice of a signed declaration as the stream holds it, and the digest of the
|
||||
// declaration it was signed over.
|
||||
func sliceOfDeclaration(data []byte, module string) ([]json.RawMessage, string, bool, error) {
|
||||
var signed link.Signed
|
||||
if err := json.Unmarshal(data, &signed); err != nil {
|
||||
return nil, "", false, fmt.Errorf("the last declaration is not readable: %w", err)
|
||||
}
|
||||
var envelope struct {
|
||||
Resources []json.RawMessage `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(signed.Declaration, &envelope); err != nil {
|
||||
return nil, "", false, fmt.Errorf("the last declaration is not readable: %w", err)
|
||||
}
|
||||
sum := sha256.Sum256(signed.Declaration)
|
||||
return sliceOfModule(envelope.Resources, module), "sha256:" + hex.EncodeToString(sum[:]), true, nil
|
||||
}
|
||||
|
||||
// sliceOfModule is, of a declaration's resources in their order, those of module: saying it, or with no module said
|
||||
// and an id that names it (sdkdecl.ModuleGuess).
|
||||
func sliceOfModule(resources []json.RawMessage, module string) []json.RawMessage {
|
||||
var out []json.RawMessage
|
||||
for _, r := range resources {
|
||||
var head struct {
|
||||
ID string `json:"id"`
|
||||
Module string `json:"module"`
|
||||
}
|
||||
if json.Unmarshal(r, &head) != nil {
|
||||
continue
|
||||
}
|
||||
owner := head.Module
|
||||
if owner == "" {
|
||||
owner, _ = sdkdecl.ModuleGuess(head.ID)
|
||||
}
|
||||
if owner == module {
|
||||
out = append(out, r)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user