mesh-sdk/go/delivery/buildrun/heardends.go

package buildrun

import (
	"context"
	"fmt"
	"sync"

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

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// **A step's end is never only its own watch's to hear** (the lost check of 2026-10-28: mesh-catalog #553's verdict
// stayed running after its step mvsgthyjed5786 had passed on shanks at 19:09:31Z; the watch of that one key, opened at
// 18:46:05Z, heard neither the step's start nor its end and never closed, while a watch opened at 18:58:48Z heard its own
// step end). HeardEnds stands between the asker (Asker) and the watch of each ask's step key: when it is
// nudged, every ask it waits on reads its step keys now (a read, not the watch) and hands what it reads to its wait, so a
// step that ended is judged even when its own watch went deaf. A watch that closes under a wait is opened again, at most
// reopenTries times. It is nudged on an event (Nudge): any step of the same build module ending (one watch of each steps
// state, HearEnds), a reconnect to the bus, and whatever else the asker hears (the checker: checker_ask_again).
// It holds no timer. Every asker of the build seats waits through it (the checker and the builder, HearEnds).

// StepReader reads, now, every step a build module's steps state holds for an ask: its keys `*.<ask id>`.
type StepReader interface {
	AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error)
}

// reopenTries is how often a wait opens its ask's watch again after it closed, before the wait ends failed.
const reopenTries = 3

// HeardEnds is an AskWaiter over Inner that also reads an ask's steps when nudged. Safe for concurrent use.
type HeardEnds struct {
	Inner AskWaiter
	Read  StepReader
	Logf  func(format string, args ...any)

	mu      sync.Mutex
	waiting map[string]*waitedAsk // by ask id
}

type waitedAsk struct {
	module string
	nudge  chan struct{}
}

func (h *HeardEnds) logf(format string, args ...any) {
	if h.Logf != nil {
		h.Logf(format, args...)
	}
}

// WatchAsk is AskWaiter's: what Inner's watch hands, and, at each nudge, what the ask's keys hold now.
func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
	in, err := h.Inner.WatchAsk(ctx, module, askID)
	if err != nil {
		return nil, err
	}
	w := &waitedAsk{module: module, nudge: make(chan struct{}, 1)}
	h.mu.Lock()
	if h.waiting == nil {
		h.waiting = map[string]*waitedAsk{}
	}
	h.waiting[askID] = w
	h.mu.Unlock()
	out := make(chan delivery.Step)
	go func() {
		defer close(out)
		defer func() {
			h.mu.Lock()
			if h.waiting[askID] == w {
				delete(h.waiting, askID)
			}
			h.mu.Unlock()
		}()
		send := func(s delivery.Step) bool {
			select {
			case out <- s:
				return true
			case <-ctx.Done():
				return false
			}
		}
		reopened := 0
		for {
			select {
			case <-ctx.Done():
				return
			case s, ok := <-in:
				if !ok {
					if ctx.Err() != nil || reopened >= reopenTries {
						h.logf("the watch of the ask %s's step on %s closed %d times; its wait ends", askID, module, reopened+1)
						return
					}
					reopened++
					if in, err = h.Inner.WatchAsk(ctx, module, askID); err != nil {
						h.logf("the watch of the ask %s's step on %s closed and cannot be opened again, so its wait ends: %v", askID, module, err)
						return
					}
					h.logf("the watch of the ask %s's step on %s closed; opened again (%d of %d)", askID, module, reopened, reopenTries)
					continue
				}
				if !send(s) {
					return
				}
			case <-w.nudge:
				if h.Read == nil {
					continue
				}
				steps, err := h.Read.AskSteps(ctx, module, askID)
				if err != nil {
					h.logf("the steps of the ask %s on %s cannot be read: %v", askID, module, err)
					continue
				}
				for _, s := range steps {
					if s.State != delivery.StepRunning {
						h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State)
					}
					if !send(s) {
						return
					}
				}
			}
		}
	}()
	return out, nil
}

// Nudge has every ask waited on of module read its steps now; module "" nudges every ask.
func (h *HeardEnds) Nudge(module string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, w := range h.waiting {
		if module == "" || w.module == module {
			select {
			case w.nudge <- struct{}{}:
			default: // a nudge is pending already
			}
		}
	}
}

// Waiting is how many asks are waited on now.
func (h *HeardEnds) Waiting() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.waiting)
}

// AskSteps is StepReader's: the keys `*.<ask id>` of the build module's steps state, read now.
func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) {
	pattern, err := delivery.AskStepWatch(askID)
	if err != nil {
		return nil, err
	}
	kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
	}
	lister, err := kv.ListKeysFiltered(ctx, pattern)
	if err != nil {
		return nil, err
	}
	var out []delivery.Step
	for key := range lister.Keys() {
		e, err := kv.Get(ctx, key)
		if err != nil {
			continue // deleted between the list and the read
		}
		if step, err := delivery.DecodeStep(e.Value()); err == nil {
			out = append(out, step)
		}
	}
	return out, nil
}

// HearEnds is the HeardEnds every asker of the build seats waits through: over inner (the asks' own watches, or a
// LostAskWaiter's inner), reading with asks, and nudged at each step end of each of modules (one watch of each steps state,
// updates only; one that closes is opened again, and said and left after reopenTries closes in a row that heard nothing). The asker nudges it at a reconnect too (Nudge("")).
func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules []string, logf func(string, ...any)) *HeardEnds {
	if logf == nil {
		logf = func(string, ...any) {}
	}
	h := &HeardEnds{Inner: inner, Read: asks, Logf: logf}
	for _, module := range modules {
		kv, err := asks.JS.KeyValue(ctx, delivery.StepsBucket(module))
		if err != nil {
			logf("the steps of %s cannot be read (%v): an ask whose own watch goes deaf is read only at a reconnect", module, err)
			continue
		}
		go h.follow(ctx, kv, module)
	}
	return h
}

func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module string) {
	silent := 0 // closes in a row having heard nothing
	for ctx.Err() == nil {
		w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
		if err != nil {
			h.logf("the steps of %s cannot be watched, so an ask whose own watch goes deaf is read only at a reconnect: %v", module, err)
			return
		}
		heard := false
		for e := range w.Updates() {
			if e == nil || e.Operation() != jetstream.KeyValuePut {
				continue
			}
			heard = true
			if step, err := delivery.DecodeStep(e.Value()); err == nil && step.State != delivery.StepRunning {
				h.Nudge(module)
			}
		}
		_ = w.Stop()
		if ctx.Err() != nil {
			return
		}
		h.Nudge(module) // a step may have ended while the watch was closing
		if heard {
			silent = 0
		} else if silent++; silent >= reopenTries {
			h.logf("the watch of the steps of %s closed %d times in a row having heard nothing, so it is left: an ask whose own watch goes deaf is read only at a reconnect", module, silent)
			return
		}
		h.logf("the watch of the steps of %s closed; opened again", module)
	}
}

// AskSteps is StepReader's: the last step written for the ask, if any.
func (f *FakeAsks) AskSteps(_ context.Context, _ string, askID string) ([]delivery.Step, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if s := f.steps[askID]; len(s) > 0 {
		return []delivery.Step{s[len(s)-1]}, nil
	}
	return nil, nil
}