mesh-controller/cmd/mesh-controller/declaration_sweep.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"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"
	"github.com/novox/mesh-controller/internal/link"
)

// **An object is removed a day after no node's latest declaration names it** (novox/hq ADR 0486 §3, issue 1696), so a node that is
// asleep and receives its latest declaration later still finds every object that declaration names. What each node's latest
// declaration names is read from the bus itself, the last message of that node's subject of the stream NODES: nothing else
// is kept of it. 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 latest declaration 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 node's latest declaration.
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
}

// namedByLatest is the name of every object this latest declaration of a node names. A body that cannot be read is an error, never
// "names none": an object the node still needs would be marked for removal.
func namedByLatest(node string, signed []byte) (map[string]bool, error) {
	var s link.Signed
	if err := json.Unmarshal(signed, &s); err != nil {
		return nil, fmt.Errorf("%s's latest declaration is not readable: %w", node, err)
	}
	names, err := namedObjects(s.Declaration)
	if err != nil {
		return nil, fmt.Errorf("%s's latest declaration is not readable: %w", node, err)
	}
	named := make(map[string]bool, len(names))
	for name := range names {
		named[name] = true
	}
	return named, nil
}

// latestDeclarations are the last signed declarations of the nodes, by node, as the stream NODES holds them. A node with none has
// no entry.
func latestDeclarations(ctx context.Context, conn *nats.Conn, nodes []string) (map[string][]byte, error) {
	js, err := conn.JetStream(nats.Context(ctx))
	if err != nil {
		return nil, err
	}
	out := map[string][]byte{}
	for _, node := range nodes {
		msg, err := js.GetLastMsg("NODES", link.DeclareSubject(node), nats.Context(ctx))
		if errors.Is(err, nats.ErrMsgNotFound) {
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("the latest declaration of %s: %w", node, err)
		}
		out[node] = msg.Data
	}
	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)()
	latest, err := latestDeclarations(working, conn, []string{node})
	if err != nil {
		return filesSweep{}, err
	}
	named := map[string]bool{}
	if body, ok := latest[node]; ok {
		if named, err = namedByLatest(node, body); err != nil {
			return filesSweep{}, err
		}
	}
	return sweepDeclarationFiles(working, busSweptBucket{store}, named, time.Now())
}