jschoubben
60253c82b7
A check pending redelivery is read before the let-through fetches (none is handed back); a deleted module is done for the plan's key; the queue an ask came from is let go of when it ends. Tries say they run under their own id.
385 lines
14 KiB
Go
385 lines
14 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
)
|
|
|
|
// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687 and 1688).
|
|
//
|
|
// **The layout other parts read** (ADR 0484 clause 12 has the checker read the same key; a layout shared between modules
|
|
// belongs in mesh-sdk, and this one is not there yet):
|
|
//
|
|
// - bucket `mesh-controller_plans-building` (broker.PlansBuildingBucket): one key per delivery plan in its build phase, named
|
|
// by the plan's id (`plan-<digits>`), value a PlanBuilding as JSON. Written by the controller (lease holder) alone: put when
|
|
// the plan's first build is asked, put again about every thirty seconds, deleted when the phase ends; the bucket's age
|
|
// (two minutes) removes the key of a controller that stopped. A reader lists the keys and needs no value to decide.
|
|
// - key `let-through` in bucket `mesh-controller_checks-running` (broker.CheckRunningBucket), value a LetThrough as JSON:
|
|
// created by the holder that takes a check which waited LetThroughAfter, refreshed at each heartbeat of that check, deleted
|
|
// when it ends. At most one such check runs at a time on the build seat, because the create fails while the key is there.
|
|
|
|
// LetThroughAfter is how long a check waits, while a delivery builds, before it is let through (ADR 0483 clause 3).
|
|
const LetThroughAfter = 30 * time.Minute
|
|
|
|
// LetThroughKey is the key of the one check let through, in the bucket of the checks running. It has no dot, so it is no
|
|
// pull request's key (`<owner>.<repo>.<number>`) and no group's (`group.<name>`).
|
|
const LetThroughKey = "let-through"
|
|
|
|
// PlanBuilding is the value of a plan's key in the bucket of the plans building.
|
|
type PlanBuilding struct {
|
|
Plan string `json:"plan"`
|
|
Since time.Time `json:"since"`
|
|
Seen time.Time `json:"seen"`
|
|
}
|
|
|
|
// LetThrough is the value of the let-through key: the builder that runs the check and when it took it.
|
|
type LetThrough struct {
|
|
Builder string `json:"builder"`
|
|
Since time.Time `json:"since"`
|
|
}
|
|
|
|
// checkGate is what a taking loop asks of the bus to hold checks while a delivery builds; a variable of the machine so a test
|
|
// hands out against fakes.
|
|
type checkGate interface {
|
|
// Building is the ids of the plans in their build phase: the keys of the bucket. An error is a bucket that cannot be read.
|
|
Building(ctx context.Context) ([]string, error)
|
|
// OldestWaiting is when the first waiting ask of a check queue was asked, read without taking it.
|
|
OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error)
|
|
// TakeLetThrough creates the let-through key for a builder: false when it is there (another builder runs a check let through).
|
|
TakeLetThrough(ctx context.Context, builder string) (bool, error)
|
|
// KeepLetThrough puts the key again, and DropLetThrough deletes it.
|
|
KeepLetThrough(ctx context.Context, builder string) error
|
|
DropLetThrough(ctx context.Context) error
|
|
}
|
|
|
|
// busGate is the gate over a connection.
|
|
type busGate struct{ conn *nats.Conn }
|
|
|
|
func (g busGate) Building(ctx context.Context) ([]string, error) {
|
|
return BuildingPlans(ctx, g.conn)
|
|
}
|
|
|
|
// BuildingPlans is the ids of the delivery plans in their build phase, from the controller's bucket. A reader that is not the
|
|
// build seat's (the checker, ADR 0484 clause 12) calls this. A bucket that cannot be read is an error; none is none building.
|
|
func BuildingPlans(ctx context.Context, conn *nats.Conn) ([]string, error) {
|
|
api, err := jetstream.New(conn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the plans building: %w", err)
|
|
}
|
|
keys, err := kv.Keys(ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the plans building: %w", err)
|
|
}
|
|
return keys, nil
|
|
}
|
|
|
|
// OldestWaiting is when the check a fetch of this queue would return was asked, read without taking anything.
|
|
//
|
|
// **It sees redeliveries** (review of issue 1688). A work queue keeps only what is not acknowledged: what the worker handed out
|
|
// (sequence up to its delivered mark) is running or waiting to be handed out again, and a fetch returns those before anything
|
|
// new. An ask there that is not said running (a key of the checks running names its build) is such a redelivery. If one is
|
|
// younger than the bound, the queue's next fetch would return it, and a young check is never let through: the queue is not a
|
|
// candidate, and nothing is fetched, so nothing is handed back and no delivery is spent. Otherwise the candidate is the oldest
|
|
// of the redeliveries and the first ask after the delivered mark.
|
|
func (g busGate) OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) {
|
|
info, err := jsOf(g.conn).ConsumerInfo(q.worker.Stream, q.worker.Name, nats.Context(ctx))
|
|
if err != nil {
|
|
return time.Time{}, false, fmt.Errorf("the %s worker: %w", q.kind, err)
|
|
}
|
|
var oldest time.Time
|
|
note := func(at time.Time) {
|
|
if oldest.IsZero() || at.Before(oldest) {
|
|
oldest = at
|
|
}
|
|
}
|
|
for seq := uint64(1); seq <= info.Delivered.Stream; {
|
|
data, stored, got, err := g.directGet(ctx, q, seq)
|
|
if err != nil {
|
|
return time.Time{}, false, err
|
|
}
|
|
if got == 0 || got > info.Delivered.Stream {
|
|
break
|
|
}
|
|
seq = got + 1
|
|
var r BuildRequest
|
|
if json.Unmarshal(data, &r) != nil {
|
|
continue
|
|
}
|
|
running, ok, err := ReadRunning(ctx, g.conn, r)
|
|
if err != nil {
|
|
return time.Time{}, false, fmt.Errorf("the checks running: %w", err)
|
|
}
|
|
if ok && running.Build == r.ID {
|
|
continue // running now
|
|
}
|
|
at := askedAtOf(data, stored)
|
|
if time.Since(at) < LetThroughAfter {
|
|
return time.Time{}, false, nil // a young redelivery is next in line
|
|
}
|
|
note(at)
|
|
}
|
|
data, stored, got, err := g.directGet(ctx, q, info.Delivered.Stream+1)
|
|
if err != nil {
|
|
return time.Time{}, false, err
|
|
}
|
|
if got != 0 {
|
|
note(askedAtOf(data, stored))
|
|
}
|
|
return oldest, !oldest.IsZero(), nil
|
|
}
|
|
|
|
// directGet reads the first ask at or after a sequence of a check queue's stream, without taking it: its body, the time the
|
|
// stream stored it, and its sequence (0 when there is none).
|
|
func (g busGate) directGet(ctx context.Context, q *takeQueue, seq uint64) ([]byte, string, uint64, error) {
|
|
body, _ := json.Marshal(map[string]any{"seq": seq, "next_by_subj": q.worker.Filters[0]})
|
|
reply, err := g.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+q.worker.Stream, body)
|
|
if err != nil {
|
|
return nil, "", 0, fmt.Errorf("the asks of the %s queue: %w", q.kind, err)
|
|
}
|
|
if st := reply.Header.Get("Status"); st == "404" || st == "408" {
|
|
return nil, "", 0, nil
|
|
}
|
|
got, _ := strconv.ParseUint(reply.Header.Get("Nats-Sequence"), 10, 64)
|
|
return reply.Data, reply.Header.Get("Nats-Time-Stamp"), got, nil
|
|
}
|
|
|
|
// askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says
|
|
// neither is taken as asked now, so it is never let through on a guess.
|
|
func askedAtOf(data []byte, stored string) time.Time {
|
|
var r BuildRequest
|
|
if json.Unmarshal(data, &r) == nil {
|
|
if at, ok := BuildAskedAt(r.ID); ok {
|
|
return at
|
|
}
|
|
}
|
|
if at, err := time.Parse(time.RFC3339Nano, stored); err == nil {
|
|
return at
|
|
}
|
|
return time.Now()
|
|
}
|
|
|
|
func jsOf(conn *nats.Conn) nats.JetStreamContext {
|
|
js, _ := conn.JetStream()
|
|
return js
|
|
}
|
|
|
|
func (g busGate) TakeLetThrough(ctx context.Context, builder string) (bool, error) {
|
|
kv, err := runningBucket(ctx, g.conn)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
|
|
if _, err := kv.Create(ctx, LetThroughKey, body); err != nil {
|
|
if errors.Is(err, jetstream.ErrKeyExists) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (g busGate) KeepLetThrough(ctx context.Context, builder string) error {
|
|
kv, err := runningBucket(ctx, g.conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
|
|
_, err = kv.Put(ctx, LetThroughKey, body)
|
|
return err
|
|
}
|
|
|
|
func (g busGate) DropLetThrough(ctx context.Context) error {
|
|
kv, err := runningBucket(ctx, g.conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := kv.Delete(ctx, LetThroughKey); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// CheckHold is what a reader of the queue (the controller's own verb) says of why checks wait: the plans building and the check
|
|
// let through now, if one.
|
|
type CheckHold struct {
|
|
Plans []string
|
|
LetThrough *LetThrough
|
|
}
|
|
|
|
// ReadCheckHold reads the plans building and the let-through key. A bucket that cannot be read is an error the caller says.
|
|
func ReadCheckHold(ctx context.Context, conn *nats.Conn) (CheckHold, error) {
|
|
var h CheckHold
|
|
plans, err := BuildingPlans(ctx, conn)
|
|
if err != nil {
|
|
return h, err
|
|
}
|
|
h.Plans = plans
|
|
kv, err := runningBucket(ctx, conn)
|
|
if err != nil {
|
|
return h, err
|
|
}
|
|
e, err := kv.Get(ctx, LetThroughKey)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return h, nil
|
|
}
|
|
if err != nil {
|
|
return h, err
|
|
}
|
|
var l LetThrough
|
|
if json.Unmarshal(e.Value(), &l) == nil {
|
|
h.LetThrough = &l
|
|
}
|
|
return h, nil
|
|
}
|
|
|
|
// gateOf is the machine's gate: the one a test set, or the bus's.
|
|
func (m *natsMachine) gateOf() checkGate {
|
|
if m.gate != nil {
|
|
return m.gate
|
|
}
|
|
return busGate{conn: m.js.Conn()}
|
|
}
|
|
|
|
// sayOnce says a fault of the hold the first time it happens and again once it has recovered and happened anew, so a bucket
|
|
// that cannot be read is in the log once and not every second.
|
|
func (m *natsMachine) sayOnce(kind string, err error) {
|
|
if m.said == nil {
|
|
m.said = map[string]bool{}
|
|
}
|
|
if err == nil {
|
|
delete(m.said, kind)
|
|
return
|
|
}
|
|
if !m.said[kind] {
|
|
m.said[kind] = true
|
|
fmt.Fprintf(os.Stderr, "%s; read as empty, so no check is held for it: %v\n", kind, err)
|
|
}
|
|
}
|
|
|
|
// building says whether a delivery builds: any key in the bucket of the plans building. **A bucket that cannot be read is said
|
|
// and read as empty** (ADR 0483 clause 2): a check wrongly held costs a pull request its turn, a fault that held every check
|
|
// would stop the mesh checking.
|
|
func (m *natsMachine) building(ctx context.Context) []string {
|
|
plans, err := m.gateOf().Building(ctx)
|
|
m.sayOnce("the bucket of the plans building cannot be read", err)
|
|
if err != nil {
|
|
// empty-on-error: ADR 0483 clause 2 — a bucket that cannot be read is said (sayOnce) and read as empty, because a check wrongly held costs a pull request its turn and a fault that held every check would stop the mesh checking.
|
|
return nil
|
|
}
|
|
return plans
|
|
}
|
|
|
|
// letOneThrough takes the oldest waiting check of the two check queues once it waited LetThroughAfter, by the one let-through
|
|
// key: nil when none waited that long, when another holder runs the check let through, or when the check was taken by
|
|
// another holder in the moment. The caller owns the key while it returns a message and releases it with dropLetThrough.
|
|
func (m *natsMachine) letOneThrough(ctx context.Context) (*nats.Msg, *takeQueue) {
|
|
gate := m.gateOf()
|
|
var oldest *takeQueue
|
|
var oldestAt time.Time
|
|
for _, q := range m.subs[1:] {
|
|
if !q.bound {
|
|
continue
|
|
}
|
|
at, found, err := gate.OldestWaiting(ctx, q)
|
|
m.sayOnce("the first waiting check of the "+q.kind+" queue cannot be read", err)
|
|
if err != nil || !found {
|
|
continue
|
|
}
|
|
if oldest == nil || at.Before(oldestAt) {
|
|
oldest, oldestAt = q, at
|
|
}
|
|
}
|
|
if oldest == nil || time.Since(oldestAt) < m.letThroughAfter() {
|
|
return nil, nil
|
|
}
|
|
created, err := gate.TakeLetThrough(ctx, m.on)
|
|
m.sayOnce("the let-through key cannot be taken", err)
|
|
if err != nil || !created {
|
|
return nil, nil
|
|
}
|
|
trying, end := context.WithTimeout(ctx, checkTakeWait)
|
|
defer end()
|
|
fetched, err := oldest.pull.Fetch(1, nats.Context(trying))
|
|
if len(fetched) == 0 {
|
|
if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.Canceled) {
|
|
m.sayOnce("the "+oldest.kind+" check queue cannot be taken from", err)
|
|
}
|
|
m.dropLetThrough(ctx)
|
|
return nil, nil
|
|
}
|
|
// Another holder took the one that was oldest in the moment between the read and the fetch: what came is a younger check,
|
|
// handed back and the key released, never run during a delivery.
|
|
if askedAtOf(fetched[0].Data, "").Add(m.letThroughAfter()).After(time.Now()) {
|
|
_ = fetched[0].NakWithDelay(time.Minute) // a delay keeps it off the queue of redeliveries a fetch reads first
|
|
m.dropLetThrough(ctx)
|
|
return nil, nil
|
|
}
|
|
m.letKey = true
|
|
return fetched[0], oldest
|
|
}
|
|
|
|
func (m *natsMachine) letThroughAfter() time.Duration {
|
|
if m.opts.LetThroughAfter > 0 {
|
|
return m.opts.LetThroughAfter
|
|
}
|
|
return LetThroughAfter
|
|
}
|
|
|
|
// dropLetThrough deletes the let-through key this holder created.
|
|
func (m *natsMachine) dropLetThrough(ctx context.Context) {
|
|
m.letKey = false
|
|
if err := m.gateOf().DropLetThrough(context.WithoutCancel(ctx)); err != nil {
|
|
fmt.Fprintf(os.Stderr, "could not take the let-through key away; it ages out: %v\n", err)
|
|
}
|
|
}
|
|
|
|
// keepLetThrough refreshes the key at every heartbeat while the check it was created for runs, and returns the end of that: it
|
|
// stops the refreshing and deletes the key.
|
|
func (m *natsMachine) keepLetThrough(ctx context.Context, every time.Duration) (end func()) {
|
|
done := make(chan struct{})
|
|
finished := make(chan struct{})
|
|
go func() {
|
|
defer close(finished)
|
|
tick := time.NewTicker(every)
|
|
defer tick.Stop()
|
|
for {
|
|
select {
|
|
case <-done:
|
|
return
|
|
case <-tick.C:
|
|
putting, stop := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
if err := m.gateOf().KeepLetThrough(putting, m.on); err != nil {
|
|
fmt.Fprintf(os.Stderr, "could not keep the let-through key: %v\n", err)
|
|
}
|
|
stop()
|
|
}
|
|
}
|
|
}()
|
|
return func() {
|
|
close(done)
|
|
<-finished
|
|
m.dropLetThrough(ctx)
|
|
}
|
|
}
|
|
|
|
// whyHeld words the plans a check is held for, for the log.
|
|
func whyHeld(plans []string) string { return strings.Join(plans, ", ") }
|