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 := chain.RequestKey(a.ID)
	// An outcome under a module that came first has no request beside it: the request is kept on it, and its own record says
	// it was answered (issue 1121).
	if ptr, found, err := c.Builds.GetRaw(ctx, chain.IDKey(a.ID)); err != nil {
		return err
	} else if found && string(ptr.Value) != chain.RequestPrefix {
		return i.keepAnsweredRequest(ctx, c, string(ptr.Value), a)
	}
	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)
}

// keepAnsweredRequest puts a request on the outcome that answered it before the request was kept, and keeps the request's
// own record as answered.
func (i *Inventory) keepAnsweredRequest(ctx context.Context, c *chain.Chain, module string, a BuildRequest) error {
	outKey := chain.BuildKey(module, a.ID)
	for try := 0; try < casTries; try++ {
		out, found, err := c.Builds.Get(ctx, outKey)
		if err != nil || !found {
			return err
		}
		if out.Value.Request == nil {
			out.Value.Request = toRequest(a)
			if _, err := c.Builds.Update(ctx, outKey, out.Value, out.Revision); errors.Is(err, chain.ErrMoved) {
				continue
			} else if err != nil {
				return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err)
			}
		}
		rec := askedRecord(a)
		rec.State, rec.Module, rec.Failed, rec.At = chain.BuildAnswered, module, out.Value.Failed, out.Value.At
		if _, err := c.Builds.Create(ctx, chain.RequestKey(a.ID), rec); err != nil && !errors.Is(err, chain.ErrExists) {
			return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err)
		}
		return nil
	}
	return fmt.Errorf("build request %s is not kept: 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, chain.RequestKey(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 := chain.RequestKey(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 := requestRecords(ctx, c)
	if err != nil {
		return nil, err
	}
	oldest := time.Now().Add(-KeptFor)
	var out []RequestOutcome
	for _, k := range all {
		id := strings.TrimPrefix(k.Key, chain.RequestPrefix+".")
		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
}

// requestRecords are the records under `req.`: the requests, and the builds that knew no module.
func requestRecords(ctx context.Context, c *chain.Chain) ([]chain.Kept[chain.Build], error) {
	return buildsIn(ctx, c, chain.RequestPrefix+".*")
}

// 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, on, err := i.ChainReads(ctx, KindBuilds)
	if err != nil || !on {
		return nil, err
	}
	all, err := requestRecords(ctx, c)
	if err != nil {
		return nil, err
	}
	var out []BuildRequest
	for _, k := range all {
		id := strings.TrimPrefix(k.Key, chain.RequestPrefix+".")
		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, on, err := i.ChainReads(ctx, KindBuilds)
	if err != nil || !on {
		return false, err
	}
	cur, found, err := c.Builds.Get(ctx, chain.RequestKey(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, chain.RequestKey(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, on, err := i.ChainReads(ctx, KindBuilds)
	if err != nil || !on {
		return err
	}
	return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.Reask = asked })
}