mesh-controller/internal/inventory/chain_requests.go

package inventory

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

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

// Build requests on the bus (novox/hq ADR 0408, issue 1091): a request is the build record in state asked, under the id its
// outcome will carry; the outcome takes its place and keeps the request. The bucket removes no key, so a request is not
// deleted after KeptFor: it is no longer listed.

func toRequest(a BuildRequest) *chain.Request {
	return &chain.Request{Repository: a.Repository, Seat: a.Seat, Path: strings.Trim(a.Path, "/"), Ref: a.Ref, Commit: a.Commit,
		For: a.For, NotAsked: a.NotAsked, OutcomeUnknown: a.OutcomeUnknown, At: a.At.UTC(), AtTerminal: a.AtTerminal}
}

func fromRequest(id string, r chain.Request) BuildRequest {
	return BuildRequest{ID: id, Repository: r.Repository, Seat: r.Seat, Path: r.Path, Ref: r.Ref, Commit: r.Commit, For: r.For,
		NotAsked: r.NotAsked, OutcomeUnknown: r.OutcomeUnknown, At: r.At, AtTerminal: r.AtTerminal}
}

// askedRecord is a request as the build record in state asked.
func askedRecord(a BuildRequest) chain.Build {
	r := toRequest(a)
	return chain.Build{ID: a.ID, State: chain.BuildAsked, Repository: a.Repository, Ref: a.Ref, Path: r.Path, Asked: r.At, Request: r}
}

func (i *Inventory) chainRecordBuildRequest(ctx context.Context, c *chain.Chain, a BuildRequest) error {
	if a.At.IsZero() {
		a.At = time.Now().UTC()
	}
	key := chainBuildKey(a.ID)
	for try := 0; try < casTries; try++ {
		_, err := c.Builds.Create(ctx, key, askedRecord(a))
		if err == nil {
			return nil
		}
		if !errors.Is(err, chain.ErrExists) {
			return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err)
		}
		// The key holds a value: the request is kept already, or its outcome came first and has no request beside it.
		cur, found, err := c.Builds.Get(ctx, key)
		if err != nil {
			return err
		}
		if !found || cur.Value.Request != nil {
			if !found {
				continue
			}
			return nil
		}
		cur.Value.Request = toRequest(a)
		if _, err := c.Builds.Update(ctx, key, cur.Value, cur.Revision); err == nil {
			return nil
		} else if !errors.Is(err, chain.ErrMoved) {
			return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err)
		}
	}
	return fmt.Errorf("build request %s is not kept on the bus: it kept moving under the write", a.ID)
}

func (i *Inventory) chainBuildRequestByID(ctx context.Context, c *chain.Chain, id string) (BuildRequest, bool, error) {
	k, found, err := c.Builds.Get(ctx, chainBuildKey(id))
	if err != nil || !found || k.Value.Request == nil {
		return BuildRequest{}, false, err
	}
	return fromRequest(id, *k.Value.Request), true, nil
}

// chainMarkRequest changes a request that has no outcome yet, by compare-and-set; one answered, or none kept, is left.
func (i *Inventory) chainMarkRequest(ctx context.Context, c *chain.Chain, id string, mark func(*chain.Request)) error {
	key := chainBuildKey(id)
	for try := 0; try < casTries; try++ {
		cur, found, err := c.Builds.Get(ctx, key)
		if err != nil || !found || cur.Value.State != chain.BuildAsked || cur.Value.Request == nil {
			return err
		}
		mark(cur.Value.Request)
		if _, err := c.Builds.Update(ctx, key, cur.Value, cur.Revision); err == nil {
			return nil
		} else if !errors.Is(err, chain.ErrMoved) {
			return fmt.Errorf("build request %s is not marked on the bus: %w", id, err)
		}
	}
	return fmt.Errorf("build request %s is not marked: it kept moving under the write", id)
}

// chainRequests is every request kept, newest first, each with its outcome where its build was heard. Requests older than
// KeptFor are not listed: the bucket keeps them, and no reader needs them.
func (i *Inventory) chainRequests(ctx context.Context, c *chain.Chain) ([]RequestOutcome, error) {
	all, err := c.Builds.All(ctx)
	if err != nil {
		return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err)
	}
	oldest := time.Now().Add(-KeptFor)
	var out []RequestOutcome
	for id, k := range all {
		r := k.Value.Request
		if r == nil || !r.At.After(oldest) {
			continue
		}
		a := RequestOutcome{BuildRequest: fromRequest(id, *r), HeardAt: r.At}
		if k.Value.State != chain.BuildAsked {
			a.Heard, a.Failed, a.Module, a.HeardAt = true, k.Value.Failed, k.Value.Module, k.Value.At
		}
		out = append(out, a)
	}
	sort.Slice(out, func(a, b int) bool {
		if !out[a].At.Equal(out[b].At) {
			return out[a].At.After(out[b].At)
		}
		return out[a].ID > out[b].ID
	})
	return out, nil
}

// RequestsToAskAgain are the requests the move found never handed over (issue 981 note 7 Q5), not yet asked again: each is
// asked once, and never twice.
func (i *Inventory) RequestsToAskAgain(ctx context.Context) ([]BuildRequest, error) {
	c, err := i.theChain(ctx)
	if err != nil {
		return nil, err
	}
	all, err := c.Builds.All(ctx)
	if err != nil {
		return nil, err
	}
	var out []BuildRequest
	for id, k := range all {
		if k.Value.State == chain.BuildAsked && k.Value.Request != nil && k.Value.Request.Reask == chain.ReaskPending {
			out = append(out, fromRequest(id, *k.Value.Request))
		}
	}
	sort.Slice(out, func(a, b int) bool { return out[a].ID < out[b].ID })
	return out, nil
}

// ClaimToAskAgain takes a request to ask again, by compare-and-set: true for the one process that took it. **Claimed
// before it is asked and settled after**, so a crash between loses the ask rather than making it twice.
func (i *Inventory) ClaimToAskAgain(ctx context.Context, id string) (bool, error) {
	c, err := i.theChain(ctx)
	if err != nil {
		return false, err
	}
	cur, found, err := c.Builds.Get(ctx, chainBuildKey(id))
	if err != nil || !found || cur.Value.Request == nil || cur.Value.Request.Reask != chain.ReaskPending {
		return false, err
	}
	cur.Value.Request.Reask = chain.ReaskClaimed
	if _, err := c.Builds.Update(ctx, chainBuildKey(id), cur.Value, cur.Revision); err != nil {
		if errors.Is(err, chain.ErrMoved) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// AskedAgainAs records the build id a claimed request was asked as.
func (i *Inventory) AskedAgainAs(ctx context.Context, id, asked string) error {
	c, err := i.theChain(ctx)
	if err != nil {
		return err
	}
	return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.Reask = asked })
}