mesh-sdk/go/follower.go

package stdio

import (
	"fmt"
	"os"
	"sync"
	"time"
)

// A watch's handler runs beside the runtime, never in its way (issue 774, the sweep of 2026-10-30).
//
// The runtime hands a watch's current values over one at a time and gives up on the watch when none was handed over
// within its bound (mesh-tools node-tools internal/bus/state.go StateWatch, 10 s). The bundle answered each change only
// once its handler returned, so a slow handler (mesh-provisioner's Rejudge waits on the lock its composition holds)
// counted against the bus's bound: the watch was given up, the SDK forgot its id, and the values the runtime still
// delivered to it were refused as "names no watch of this bundle", 1257 times in an hour on the control-node.
//
// So each watch has a follower: a change is taken into it in the order the runtime hands it over and answered at once,
// and the handler runs on the follower's own goroutine, in that order. Watch still returns only once every current
// value was handled. A handler's error is said on the bundle's standard error, naming the watch and the change, since
// the runtime no longer hears it. A log watch keeps its answer after the handler: its runtime sends the next entry only
// after the answer, which is the back-pressure a replay of a long log needs.

// follower runs one watch's handler on its own goroutine, in order.
type follower[T any] struct {
	what     string
	handle   func(T) error
	describe func(T) string
	mu       sync.Mutex
	items    []followed[T]
	wake     chan struct{}
}

// followed is one change, or with no change a mark that closes done once everything before it was handled.
type followed[T any] struct {
	change T
	mark   bool
	done   chan struct{}
}

func newFollower[T any](what string, handle func(T) error, describe func(T) string) *follower[T] {
	f := &follower[T]{what: what, handle: handle, describe: describe, wake: make(chan struct{}, 1)}
	go f.run()
	return f
}

// take queues a change; it never waits on the handler.
func (f *follower[T]) take(change T) {
	f.mu.Lock()
	f.items = append(f.items, followed[T]{change: change})
	f.mu.Unlock()
	f.poke()
}

// handled is closed once every change taken before it was handled.
func (f *follower[T]) handled() <-chan struct{} {
	done := make(chan struct{})
	f.mu.Lock()
	f.items = append(f.items, followed[T]{mark: true, done: done})
	f.mu.Unlock()
	f.poke()
	return done
}

func (f *follower[T]) poke() {
	select {
	case f.wake <- struct{}{}:
	default:
	}
}

func (f *follower[T]) run() {
	for range f.wake {
		for {
			f.mu.Lock()
			if len(f.items) == 0 {
				f.mu.Unlock()
				break
			}
			next := f.items[0]
			f.items = f.items[1:]
			f.mu.Unlock()
			if next.mark {
				close(next.done)
				continue
			}
			if err := f.handle(next.change); err != nil {
				sayf("%s: the handler did not take %s: %v; the next change hands it again", f.what, f.describe(next.change), err)
			}
		}
	}
}

// sayf says something the runtime cannot say for the bundle, on its standard error, which its journal keeps.
func sayf(format string, args ...any) {
	fmt.Fprintf(os.Stderr, "[mesh-sdk] "+format+"\n", args...)
}

// gaveUpMost bounds what a bundle remembers of the watches it gave up.
const gaveUpMost = 256

// gaveUp keeps why a watch was given up, so a change the runtime still hands to it is refused saying so, never as a
// watch nobody asked for. Called with c.mu held.
func (c *channel) gaveUp(id, why string) {
	if c.givenUp == nil || len(c.givenUp) >= gaveUpMost {
		c.givenUp = map[string]string{}
	}
	c.givenUp[id] = why
}

// noWatch is the refusal of a change for watch id, which this bundle never asked or gave up.
func (c *channel) noWatch(kind, id string) string {
	c.mu.Lock()
	why, given := c.givenUp[id]
	c.mu.Unlock()
	if given {
		sayf("%s: the runtime still hands over changes for the watch %s, which this bundle gave up: %s", kind, id, why)
		return fmt.Sprintf("%s: this bundle gave the watch %s up (%s); stop handing it over", kind, id, why)
	}
	return kind + " names no watch of this bundle"
}

// Follow watches key of the state as Watch does and asks again until the watch is taken, saying each give-up and
// when it asks again (2 s, doubled to a minute): the loop every module wrote for itself, in one place. It returns once
// the watch is taken; call it beside the code that needs it.
func (s KeptState) Follow(key string, handle func(StateChange) error) {
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := s.Watch(key, handle)
		if err == nil {
			return
		}
		sayf("the state %s is not watched yet (%v); asking again in %s", s.name, err, wait)
		time.Sleep(wait)
	}
}