mesh-controller/internal/link/calls_kept.go

package link

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

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

	"github.com/novox/mesh-controller/internal/broker"
)

// Calls kept on the bus (novox/hq to-be 45 §6).
//
// **Two keys a call**: `<id>` holds the record — verb, arguments as kept, caller, state, times — and
// `<id>.answer` the answer, bounded. A listing watches the records alone, so reading the last thousand
// calls reads the last thousand small records and not a thousand answers; one call asked by id reads
// both. A call's id is one token, so the record's key never holds a dot and `*` matches records only.

// BusCalls keeps calls in the controller's calls bucket.
type BusCalls struct {
	kv jetstream.KeyValue
}

// CallsOnTheBus opens the calls bucket the controller asserts at its start.
func CallsOnTheBus(ctx context.Context, conn *nats.Conn) (*BusCalls, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	kv, err := api.KeyValue(ctx, broker.CallsBucket)
	if err != nil {
		return nil, fmt.Errorf("the calls bucket %s is not on the bus — the controller asserts it at its "+
			"start, so one older than this has not: %w", broker.CallsBucket, err)
	}
	return &BusCalls{kv: kv}, nil
}

const answerKey = ".answer"

// Keep writes a call's record, and its answer when it has one.
func (b *BusCalls) Keep(ctx context.Context, c Call) error {
	answer := c.Answer
	c.Answer = nil
	if len(answer) > 0 {
		if len(answer) > broker.CallAnswerBytes {
			answer, _ = json.Marshal(map[string]any{"cut": fmt.Sprintf("an answer of %d bytes; the first %d "+
				"are kept", len(answer), broker.CallAnswerBytes), "start": string(answer[:broker.CallAnswerBytes])})
		}
		// The answer before the record, so a record saying a call finished never points at an answer
		// not yet written.
		if _, err := b.kv.Put(ctx, c.ID+answerKey, answer); err != nil {
			return err
		}
	}
	record, err := json.Marshal(c)
	if err != nil {
		return err
	}
	_, err = b.kv.Put(ctx, c.ID, record)
	return err
}

// Kept is one call, with its answer.
func (b *BusCalls) Kept(ctx context.Context, id string) (Call, bool, error) {
	entry, err := b.kv.Get(ctx, id)
	if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrInvalidKey) {
		return Call{}, false, nil
	}
	if err != nil {
		return Call{}, false, err
	}
	var c Call
	if err := json.Unmarshal(entry.Value(), &c); err != nil {
		return Call{}, false, fmt.Errorf("the record of %s on the bus is not a call: %w", id, err)
	}
	answer, err := b.kv.Get(ctx, id+answerKey)
	switch {
	case err == nil:
		c.Answer = json.RawMessage(answer.Value())
	case !errors.Is(err, jetstream.ErrKeyNotFound):
		return Call{}, false, err
	}
	return c, true, nil
}

// markAbandonedTries bounds the compare-and-set loop below (novox/hq issue 711 review round 1): a
// record that keeps changing on every read is not this sweep's to keep retrying forever over.
const markAbandonedTries = 10

// MarkAbandoned marks id abandoned unless its record already says it finished (novox/hq issue 711):
// a controller starting finds a call running under a holder that stopped, but that holder may have
// written the finish — with its kept answer — around the same moment. The record's own
// compare-and-set decides the race rather than the order these two writers happen to run in: this
// reads the record, and if it still says running, writes abandoned at that revision; a finish landing
// in between fails the write with a revision mismatch, and this rereads and finds the finish, which
// then stands. A record already resolved, or gone, is returned as found, so the caller never writes
// abandoned over it. Past markAbandonedTries attempts, it gives up and returns an error naming the
// call instead of looping forever (novox/hq issue 711 review round 1); Durably reports that as it
// reports any other keeper error.
func (b *BusCalls) MarkAbandoned(ctx context.Context, id string) (Call, error) {
	for try := 0; try < markAbandonedTries; try++ {
		entry, err := b.kv.Get(ctx, id)
		if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrInvalidKey) {
			return Call{}, nil
		}
		if err != nil {
			return Call{}, err
		}
		var c Call
		if err := json.Unmarshal(entry.Value(), &c); err != nil {
			return Call{}, fmt.Errorf("the record of %s on the bus is not a call: %w", id, err)
		}
		if c.State != CallRunning {
			// A finish (or an earlier mark) is already the record's word: it stands, untouched.
			return c, nil
		}
		c.State = CallAbandoned
		record, err := json.Marshal(c)
		if err != nil {
			return Call{}, err
		}
		if _, err := b.kv.Update(ctx, id, record, entry.Revision()); err != nil {
			if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
				continue // something — a finish, most likely — wrote since: reread and judge it again
			}
			return Call{}, err
		}
		return c, nil
	}
	return Call{}, fmt.Errorf("%s: its record kept changing; it is left as it stands", id)
}

// Recent is every call the bucket holds, newest first, without answers: read once through a watch
// of the records, which hands over the current value of each and then says it has.
func (b *BusCalls) Recent(ctx context.Context) ([]Call, error) {
	w, err := b.kv.Watch(ctx, "*", jetstream.IgnoreDeletes())
	if err != nil {
		return nil, err
	}
	defer func() { _ = w.Stop() }()
	var out []Call
	for {
		select {
		case <-ctx.Done():
			return nil, fmt.Errorf("reading the calls bucket: %w", ctx.Err())
		case entry := <-w.Updates():
			if entry == nil {
				// Every current value handed over.
				sort.SliceStable(out, func(i, j int) bool { return out[i].Started.After(out[j].Started) })
				return out, nil
			}
			var c Call
			if json.Unmarshal(entry.Value(), &c) == nil && c.ID != "" {
				out = append(out, c)
			}
		}
	}
}