jschoubben
76c48fa074
A node applies exactly the composition its want names (ADR 0539 clause 1), which may be a revision a newer one replaced: its files over 64 KiB stay until no want names it.
354 lines
12 KiB
Go
354 lines
12 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
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/inventory"
|
|
)
|
|
|
|
// **An object is removed a day after no composition of the node names it** (novox/hq ADR 0486 §3, ADR 0539 §4, issue 1696), so a
|
|
// node that is asleep and applies its wants later still finds every object they name. What a node's compositions name is read
|
|
// from its bucket `composed-<node>`, where the controller keeps each module's five newest fingerprints and the node's own: nothing
|
|
// else is kept of them. When an object is first found named by none, the time is written in its own metadata (declarationUnnamed);
|
|
// found named again, that is erased; found unnamed for a day, it is removed. Each node has a bucket of its own, judged against that
|
|
// node's compositions alone. Objects are immutable and regenerable from the controller's state, so a removal loses nothing.
|
|
|
|
// declarationUnnamed is the key of an object's metadata holding when it was first found named by no composition of the node.
|
|
const declarationUnnamed = "unnamed-since"
|
|
|
|
// sweepEvery is how often the bucket is judged: the removal is exact to within this.
|
|
const sweepEvery = time.Hour
|
|
|
|
// bucketObject is one object of the bucket as the sweep sees it.
|
|
type bucketObject struct {
|
|
Name string
|
|
// UnnamedSince is when it was first found named by none, zero when it was named at the last look.
|
|
UnnamedSince time.Time
|
|
}
|
|
|
|
// sweptBucket is what the sweep needs of the bucket.
|
|
type sweptBucket interface {
|
|
Objects(ctx context.Context) ([]bucketObject, error)
|
|
MarkUnnamed(ctx context.Context, name string, since time.Time) error
|
|
ClearUnnamed(ctx context.Context, name string) error
|
|
// Delete removes the object if it still carries the mark it was judged by, and says whether it did: a send that put it again
|
|
// since has erased the mark, and the object is left.
|
|
Delete(ctx context.Context, name string, markedAt time.Time) (bool, error)
|
|
}
|
|
|
|
// filesSweep says what a sweep did.
|
|
type filesSweep struct{ Kept, Marked, Cleared, Removed int }
|
|
|
|
// sweepDeclarationFiles judges every object against the names of the latest declarations: named ones are kept (and a
|
|
// mark of being unnamed is erased), unnamed ones are marked at the first look and removed once the mark is a day old.
|
|
func sweepDeclarationFiles(ctx context.Context, bucket sweptBucket, named map[string]bool, now time.Time) (filesSweep, error) {
|
|
var r filesSweep
|
|
objects, err := bucket.Objects(ctx)
|
|
if err != nil {
|
|
return r, err
|
|
}
|
|
sort.Slice(objects, func(i, j int) bool { return objects[i].Name < objects[j].Name })
|
|
for _, o := range objects {
|
|
switch {
|
|
case named[o.Name]:
|
|
r.Kept++
|
|
if !o.UnnamedSince.IsZero() {
|
|
if err := bucket.ClearUnnamed(ctx, o.Name); err != nil {
|
|
return r, fmt.Errorf("the object %s: %w", o.Name[:12], err)
|
|
}
|
|
r.Cleared++
|
|
}
|
|
case o.UnnamedSince.IsZero():
|
|
if err := bucket.MarkUnnamed(ctx, o.Name, now); err != nil {
|
|
return r, fmt.Errorf("the object %s: %w", o.Name[:12], err)
|
|
}
|
|
r.Marked++
|
|
case now.Sub(o.UnnamedSince) >= broker.DeclarationFilesKeptFor:
|
|
gone, err := bucket.Delete(ctx, o.Name, o.UnnamedSince)
|
|
if err != nil {
|
|
return r, fmt.Errorf("the object %s: %w", o.Name[:12], err)
|
|
}
|
|
if gone {
|
|
r.Removed++
|
|
} else {
|
|
r.Kept++
|
|
}
|
|
default:
|
|
r.Kept++
|
|
}
|
|
}
|
|
return r, nil
|
|
}
|
|
|
|
// namedByCompositions is the name of every object the compositions of a node name: the files over 64 KiB each carries by
|
|
// digest (ADR 0539 §4). A composition that cannot be read is an error, never "names none": an object the node still needs
|
|
// would be marked for removal.
|
|
func namedByCompositions(node string, values [][]byte) (map[string]bool, error) {
|
|
named := map[string]bool{}
|
|
for _, value := range values {
|
|
c, err := sdkdelivery.DecodeComposed(value)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("a composition of %s is not readable: %w", node, err)
|
|
}
|
|
body, err := json.Marshal(map[string]any{"resources": c.Resources})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
names, err := namedObjects(body)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("a composition of %s is not readable: %w", node, err)
|
|
}
|
|
for name := range names {
|
|
named[name] = true
|
|
}
|
|
}
|
|
return named, nil
|
|
}
|
|
|
|
// keptCompositionsOf are the values of every composition in a node's bucket `composed-<node>` a node may still apply:
|
|
// each key's newest revision (each module's five newest fingerprints and the node's own, ADR 0539 §4), and any older
|
|
// revision of it whose digest a want of the node names, since the node applies exactly the composition its want names
|
|
// (clause 1) and a want may still name a revision a newer one replaced (the review's C1). A node with no bucket has none.
|
|
func keptCompositionsOf(ctx context.Context, js jetstream.JetStream, node string) ([][]byte, error) {
|
|
bucket, err := sdkdelivery.ComposedBucketFor(node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
kv, err := js.KeyValue(ctx, bucket)
|
|
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the compositions of %s: %w", node, err)
|
|
}
|
|
keys, err := kv.Keys(ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the compositions of %s: %w", node, err)
|
|
}
|
|
wanted, err := digestsWantedOn(ctx, js, node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out [][]byte
|
|
for _, key := range keys {
|
|
history, err := kv.History(ctx, key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the composition %s: %w", key, err)
|
|
}
|
|
out = append(out, keptRevisions(history, wanted)...)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// keptRevisions are, of one key's history oldest first, the newest value when the key is not deleted, and every older
|
|
// value whose composition's digest is wanted.
|
|
func keptRevisions(history []jetstream.KeyValueEntry, wanted map[string]bool) [][]byte {
|
|
var out [][]byte
|
|
for i, e := range history {
|
|
if e.Operation() != jetstream.KeyValuePut {
|
|
continue
|
|
}
|
|
if i == len(history)-1 {
|
|
out = append(out, e.Value())
|
|
continue
|
|
}
|
|
c, err := sdkdelivery.DecodeComposed(e.Value())
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if d, err := sdkdelivery.ComposedDigestOf(c); err == nil && wanted[d] {
|
|
out = append(out, e.Value())
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// digestsWantedOn is the digest of every composition a want of the node names, its own want's among them; none when
|
|
// the assigner's bucket is not there yet.
|
|
func digestsWantedOn(ctx context.Context, js jetstream.JetStream, node string) (map[string]bool, error) {
|
|
kv, err := js.KeyValue(ctx, sdkdelivery.AssignerWantsBucket)
|
|
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
|
return map[string]bool{}, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the wants of %s: %w", node, err)
|
|
}
|
|
lister, err := kv.ListKeys(ctx, jetstream.MetaOnly())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the wants of %s: %w", node, err)
|
|
}
|
|
out := map[string]bool{}
|
|
for key := range lister.Keys() {
|
|
e, err := kv.Get(ctx, key)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if n, _, err := sdkdelivery.ParseWantKey(key); err == nil && n == node {
|
|
if w, err := sdkdelivery.DecodeWant(e.Value()); err == nil && w.Composed != "" {
|
|
out[w.Composed] = true
|
|
}
|
|
} else if n, err := sdkdelivery.ParseWantNodeKey(key); err == nil && n == node {
|
|
if w, err := sdkdelivery.DecodeNodeWant(e.Value()); err == nil {
|
|
out[w.Composed] = true
|
|
}
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// busSweptBucket is the bucket on the controller's connection.
|
|
type busSweptBucket struct{ store jetstream.ObjectStore }
|
|
|
|
func (b busSweptBucket) Objects(ctx context.Context) ([]bucketObject, error) {
|
|
infos, err := b.store.List(ctx)
|
|
if errors.Is(err, jetstream.ErrNoObjectsFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]bucketObject, 0, len(infos))
|
|
for _, info := range infos {
|
|
o := bucketObject{Name: info.Name}
|
|
if at, err := time.Parse(time.RFC3339, info.Metadata[declarationUnnamed]); err == nil {
|
|
o.UnnamedSince = at
|
|
}
|
|
out = append(out, o)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (b busSweptBucket) meta(ctx context.Context, name string, change func(map[string]string)) error {
|
|
info, err := b.store.GetInfo(ctx, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
meta := info.ObjectMeta
|
|
kept := map[string]string{}
|
|
for k, v := range meta.Metadata {
|
|
kept[k] = v
|
|
}
|
|
change(kept)
|
|
meta.Metadata = kept
|
|
return b.store.UpdateMeta(ctx, name, meta)
|
|
}
|
|
|
|
func (b busSweptBucket) MarkUnnamed(ctx context.Context, name string, since time.Time) error {
|
|
return b.meta(ctx, name, func(m map[string]string) { m[declarationUnnamed] = since.UTC().Format(time.RFC3339) })
|
|
}
|
|
|
|
func (b busSweptBucket) ClearUnnamed(ctx context.Context, name string) error {
|
|
return b.meta(ctx, name, func(m map[string]string) { delete(m, declarationUnnamed) })
|
|
}
|
|
|
|
func (b busSweptBucket) Delete(ctx context.Context, name string, markedAt time.Time) (bool, error) {
|
|
info, err := b.store.GetInfo(ctx, name)
|
|
if errors.Is(err, jetstream.ErrObjectNotFound) {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if at, perr := time.Parse(time.RFC3339, info.Metadata[declarationUnnamed]); perr != nil || !at.Equal(markedAt) {
|
|
return false, nil
|
|
}
|
|
return true, b.store.Delete(ctx, name)
|
|
}
|
|
|
|
// sweepingDeclarationFiles judges the bucket every sweepEvery for as long as this controller serves and holds the lease.
|
|
func sweepingDeclarationFiles(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn) {
|
|
tick := time.NewTicker(sweepEvery)
|
|
defer tick.Stop()
|
|
failing := ""
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-tick.C:
|
|
}
|
|
if epoch, err := theLease.epoch(ctx); err != nil || epoch == 0 {
|
|
continue
|
|
}
|
|
why := sweepOnce(ctx, inv, conn)
|
|
if why != failing {
|
|
if why != "" {
|
|
fmt.Printf("the bucket of declaration files cannot be swept: %s\n", why)
|
|
} else {
|
|
fmt.Printf("the bucket of declaration files is swept again\n")
|
|
}
|
|
failing = why
|
|
}
|
|
}
|
|
}
|
|
|
|
// sweepOnce judges the bucket of every node against that node's latest declaration, one bucket at a time under the node's lock, with
|
|
// the declaration read again under it: a send that put an object and published its declaration is seen, and one that has not yet
|
|
// is waiting for the lock.
|
|
func sweepOnce(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn) string {
|
|
nodes, err := inv.Nodes(ctx)
|
|
if err != nil {
|
|
return fmt.Sprintf("the nodes cannot be read: %v", err)
|
|
}
|
|
var problems []string
|
|
var total filesSweep
|
|
for _, n := range nodes {
|
|
r, err := sweepNode(ctx, conn, n.Name)
|
|
total.Kept, total.Marked, total.Cleared, total.Removed = total.Kept+r.Kept, total.Marked+r.Marked, total.Cleared+r.Cleared, total.Removed+r.Removed
|
|
if err != nil {
|
|
problems = append(problems, fmt.Sprintf("%s: %v", n.Name, err))
|
|
}
|
|
}
|
|
if total.Removed+total.Marked+total.Cleared > 0 {
|
|
fmt.Printf("the buckets of declaration files: %d object(s) kept, %d found unnamed, %d named again, %d removed\n",
|
|
total.Kept, total.Marked, total.Cleared, total.Removed)
|
|
}
|
|
return strings.Join(problems, "; ")
|
|
}
|
|
|
|
func sweepNode(ctx context.Context, conn *nats.Conn, node string) (filesSweep, error) {
|
|
working, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
|
defer cancel()
|
|
js, err := jetstream.New(conn)
|
|
if err != nil {
|
|
return filesSweep{}, err
|
|
}
|
|
bucket, err := broker.DeclarationFilesBucketOf(node)
|
|
if err != nil {
|
|
return filesSweep{}, fmt.Errorf("the bucket of node %q: %w", node, err)
|
|
}
|
|
store, err := js.ObjectStore(working, bucket)
|
|
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
|
return filesSweep{}, nil // a node never sent a reference has no bucket yet
|
|
}
|
|
if err != nil {
|
|
return filesSweep{}, fmt.Errorf("the bucket cannot be read: %w", err)
|
|
}
|
|
defer lockFilesOf(node)()
|
|
values, err := keptCompositionsOf(working, js, node)
|
|
if err != nil {
|
|
return filesSweep{}, err
|
|
}
|
|
named, err := namedByCompositions(node, values)
|
|
if err != nil {
|
|
return filesSweep{}, err
|
|
}
|
|
return sweepDeclarationFiles(working, busSweptBucket{store}, named, time.Now())
|
|
}
|