mesh-delivery/modules/mesh-delivery-checker/internal/check/awaiting.go

package check

import (
	"context"
	"errors"
	"fmt"

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

// ErrAssignerNotRunning is what a wait for a composition answers when no instance of the assigner is online: nobody is
// making it, so the check does not wait for it (issue 1297). The verdict says the composition is not judged, and when the
// assigner starts and writes it, the checker takes it in then (Checker.Composed).
var ErrAssignerNotRunning = errors.New(delivery.AssignerNotRunning)

// PresenceEvent is a change of the watchman's presence bucket, or Settled once what stands was handed.
type PresenceEvent struct {
	Key     string
	Online  bool
	Settled bool
}

// CompositionEvent is a change of the composition's key (its value, nil for a delete), or Settled once what stands was
// handed.
type CompositionEvent struct {
	Value   []byte
	Settled bool
}

// AwaitComposition answers the assigner's composition once one is written that does not wait, while an instance of the
// assigner is online by the watchman's presence (`part.mesh-delivery-assigner.<instance>`). Once both watches handed
// what stands, no instance online is ErrAssignerNotRunning at once, and so is the last instance going offline during the
// wait. No timer: every answer follows an event; ctx ending is ctx's error.
func AwaitComposition(ctx context.Context, presence <-chan PresenceEvent, comps <-chan CompositionEvent) (delivery.CompositionVerdict, error) {
	online := map[string]bool{}
	presenceSettled, compSettled := false, false
	for {
		select {
		case <-ctx.Done():
			return delivery.CompositionVerdict{}, ctx.Err()
		case e, ok := <-presence:
			if !ok {
				return delivery.CompositionVerdict{}, errors.New("the watch of the assigner's presence ended")
			}
			if e.Settled {
				presenceSettled = true
				break
			}
			if _, part, instance, err := delivery.ParsePresenceKey(e.Key); err == nil && part == delivery.AssignerModule {
				if e.Online {
					online[instance] = true
				} else {
					delete(online, instance)
				}
			}
		case e, ok := <-comps:
			if !ok {
				return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended")
			}
			if e.Settled {
				compSettled = true
				break
			}
			if e.Value == nil {
				break
			}
			v, err := delivery.DecodeCompositionVerdict(e.Value)
			if err != nil {
				return delivery.CompositionVerdict{}, fmt.Errorf("the composition cannot be read: %w", err)
			}
			if v.State != delivery.CompositionWaits {
				return v, nil
			}
		}
		if presenceSettled && compSettled && len(online) == 0 {
			return delivery.CompositionVerdict{}, ErrAssignerNotRunning
		}
	}
}

// withComposition is v with the assigner's composition taken in: a failed one fails the check, naming its machine.
func withComposition(v *delivery.Verdict, comp delivery.CompositionVerdict) {
	v.Composition = comp.Composition()
	v.Took.Composition = comp.Took
	if comp.State == delivery.CompositionFail {
		machine := comp.Machine
		if machine == "" {
			machine = "the composition"
		}
		v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words}
	}
}

// notJudged says whether a verdict's composition was not judged because the assigner was not running.
func notJudged(v delivery.Verdict) bool {
	return v.Composition != nil && v.Composition.State == delivery.CompositionNotRun && v.Composition.Words == delivery.AssignerNotRunning
}

// Composed takes in a composition the assigner wrote after the check's verdict said it was not judged because the
// assigner was not running (the assigner started since): the verdict and its status are written again with it. A
// composition of heads no candidate holds now, or of a verdict that judged its composition, is left.
func (ch *Checker) Composed(ctx context.Context, comp delivery.CompositionVerdict) {
	ch.init()
	if comp.State == delivery.CompositionWaits {
		return
	}
	ckey, err := delivery.CandidatesKey(comp.Group)
	if err != nil {
		return
	}
	ch.mu.Lock()
	c, ok := ch.known[ckey]
	ch.mu.Unlock()
	if !ok {
		return
	}
	cs := CaseOf(c, ch.heads())
	if cs == nil || cs.HeadsHash != comp.Heads {
		return
	}
	cur, rev, found, err := ch.Verdicts.Get(ctx, cs.Key)
	if err != nil || !found {
		return
	}
	v, err := delivery.DecodeVerdict(cur)
	if err != nil || v.State != delivery.VerdictPass || !notJudged(v) {
		return
	}
	withComposition(&v, comp)
	data, err := delivery.EncodeVerdict(v)
	if err != nil {
		ch.logf("%s: the verdict with its composition cannot be written: %v", cs.Key, err)
		return
	}
	if _, err := ch.Verdicts.Update(ctx, cs.Key, data, rev); err != nil {
		ch.logf("%s: the verdict with its composition could not be written: %v", cs.Key, err)
		return
	}
	ch.writeOwed(ctx, cs, v, Outcome{Verdict: v})
	ch.logf("%s: the composition came after the verdict: %s", cs.Key, comp.State)
}