mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/oldform.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms"

	"mesh-delivery-builder/internal/versions"
)

// oldForm keeps the old reference form's window (design 51 *Every build keeps the old reference form*, DLV-B9) in the
// state old-form: stepped on every change of what the counted nodes state, of the assignments, on a hand drop and at each
// build key written, never on a timer. A node holding it seven days is the finding old-form-held; its close is said once
// as version.old-form-closed and in the builder's log.
type oldForm struct {
	forms    forms.ArtifactForms
	findings *findings
	// said emits version.old-form-closed.
	said func(versions.OldFormClosedBody)
	// log writes the close into the builder's build log (build.said); nil writes none.
	log  func(builds.BuildLogEntry)
	logf func(string, ...any)
	now  func() time.Time

	// bounds is the builder's state bounds: each node holding the window has a key there whose time to live ends at its
	// seven days, so old-form-held is raised then, not at the next change (review round 1); nil keeps none.
	bounds func() jetstream.KeyValue

	mu sync.Mutex
	w  versions.OldFormWindow
	kv jetstream.KeyValue
	// overdue are the nodes the finding is raised for now.
	overdue []string
}

// bind takes the kept window in, so a restarted builder keeps each node's since and the nodes dropped by hand, and keeps
// every later step in kv.
func (o *oldForm) bind(ctx context.Context, kv jetstream.KeyValue) {
	e, err := kv.Get(ctx, versions.OldFormKey)
	o.mu.Lock()
	defer o.mu.Unlock()
	switch {
	case errors.Is(err, jetstream.ErrKeyNotFound):
	case err != nil:
		o.logf("%s.%s cannot be read, so the window starts again: %v", versions.OldFormState, versions.OldFormKey, err)
	default:
		w, err := versions.DecodeOldFormWindow(e.Value())
		if err != nil {
			o.logf("%v; the window starts again", err)
			break
		}
		// Taken in under what was stepped before the state was bound: the kept since and drops are the older truth.
		for n, s := range o.w.Since {
			if _, ok := w.Since[n]; !ok && w.Open && slices.Contains(w.Holding, n) {
				w.Since[n] = s
			}
		}
		for _, d := range o.w.Dropped {
			w, _ = w.Drop(d.Node, d.By, d.Why, d.At)
		}
		o.w = w
	}
	o.kv = kv
	o.keep(ctx)
}

// dropped is the nodes dropped by hand.
func (o *oldForm) dropped() map[string]bool {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.w.DroppedSet()
}

// step moves the window to what counted states.
// named is every node the read named (the assignments's nodes before any is dropped), counted the statements of those
// still counted: a read naming no node judges nothing (window.StepWindow), so it never closes the window.
func (o *oldForm) step(ctx context.Context, named []string, counted map[string][]string) {
	o.mu.Lock()
	defer o.mu.Unlock()
	before := o.w
	next, closed, overdue := o.w.Step(named, versions.Holding(o.forms, counted, o.w.DroppedSet()), o.now())
	o.w = next
	if !slices.Equal(before.Holding, next.Holding) || before.Open != next.Open || closed {
		o.keep(ctx)
	}
	for _, n := range o.overdue {
		if !slices.Contains(overdue, n) {
			o.findings.clear(ctx, versions.OldFormHeld, meshconditionsclient.ScopeMachine, n)
		}
	}
	for _, n := range overdue {
		o.findings.raise(ctx, finding{kind: versions.OldFormHeld, scope: meshconditionsclient.ScopeMachine, id: n,
			summary: fmt.Sprintf("the node-engine of %s has read only an older reference form since %s, so every build key still carries it",
				n, next.Since[n].Format(time.RFC3339)),
			headline:    "A node holds the old reference form",
			explanation: "The builder writes an older reference form into every build key while a counted node's engine reads none of the newest (ADR 0624 §1); this node has held it seven days.",
			resolved:    "The node's engine states the newest form, the node leaves the assignments, or it is dropped as retired.",
			needs:       "builder_old_form shows the window; a retired node is dropped there by hand; builder_engine_reads asks the node again."})
	}
	o.overdue = overdue
	o.keepBounds(ctx, before, next, overdue)
	if closed {
		body := next.Closing()
		line := fmt.Sprintf("the old reference form's window closed at %s: no counted node-engine reads only an older form (last held by %s; dropped by hand: %d); new build keys leave it out",
			body.At.Format(time.RFC3339), strings.Join(body.LastHolding, ", "), len(body.Dropped))
		o.logf("%s", line)
		// In the build log too (design 51 *The node-engine's own build*), under the builder's own key, the window's key
		// standing for the fingerprint: the close is no one build's.
		if o.log != nil {
			o.log(builds.BuildLogEntry{Event: builds.BuildSaidEvent, Module: builds.BuilderModule, FP: versions.OldFormKey,
				At: body.At.UTC(), Line: line, Why: versions.SaidOldFormClosed})
		}
		if o.said != nil {
			o.said(body)
		}
	}
}

// keepBounds writes a bound for each node holding the window and not yet overdue, ending at its seven days, and deletes
// the bound of each node that no longer holds it. A standing bound keeps its time (a Create). Called with mu held.
func (o *oldForm) keepBounds(ctx context.Context, before, next versions.OldFormWindow, overdue []string) {
	if o.bounds == nil {
		return
	}
	kv := o.bounds()
	if kv == nil {
		return
	}
	for _, n := range before.Holding {
		if !slices.Contains(next.Holding, n) {
			if err := kv.Delete(ctx, versions.OldFormBoundKey(n)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
				o.logf("the old form's bound of %s is not deleted: %v", n, err)
			}
		}
	}
	for _, n := range next.Holding {
		left := next.Since[n].Add(versions.FindingAfter).Sub(o.now())
		if slices.Contains(overdue, n) || left <= 0 {
			continue
		}
		if left < time.Second {
			left = time.Second // the bus's least time to live
		}
		if _, err := kv.Create(ctx, versions.OldFormBoundKey(n), versions.EncodeBound("the seven days "+n+" holds the old form's window", o.now().Add(left)), jetstream.KeyTTL(left)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
			o.logf("the old form's bound of %s is not written, so its seven days are seen at the next change only: %v", n, err)
		}
	}
}

// drop is builder_old_form's drop: the node leaves the count, the window is stepped by the caller.
func (o *oldForm) drop(ctx context.Context, node, by, why string, at time.Time) (string, error) {
	o.mu.Lock()
	defer o.mu.Unlock()
	w, err := o.w.Drop(node, by, why, at)
	if err != nil {
		return "", err
	}
	o.w = w
	o.keep(ctx)
	return node + " dropped from the old form's count by hand; the window is stepped now", nil
}

// keep writes the window, once the state is bound; until then it is kept here only. Called with mu held.
func (o *oldForm) keep(ctx context.Context) {
	if o.kv == nil {
		return
	}
	raw, err := json.Marshal(o.w)
	if err == nil {
		_, err = o.kv.Put(ctx, versions.OldFormKey, raw)
	}
	if err != nil {
		o.logf("%s.%s cannot be written: %v", versions.OldFormState, versions.OldFormKey, err)
	}
}