Keep a build request as the build in state asked (issue 1091)
A request is the build's record under the id its outcome will carry, in state asked; the outcome takes its place and keeps the request beside it, so assign, the pending assignments and the planner read the builds bucket alone. The bucket removes no key, so a request past thirty days is not listed instead of deleted. The move folds build_request into the same records (one kind, builds, counted against both tables) and marks the requests never handed over to be asked again once: claimed before it is asked and settled after, so a crash loses the ask rather than making it twice. Also tests the chain buckets' options on a real bus, the assignments' mirror options included.
This commit is contained in:
@@ -52,6 +52,7 @@ func movingTheChain(ctx context.Context, inv *inventory.Inventory) {
|
||||
switch {
|
||||
case err == nil:
|
||||
go clearChainMove(ctx)
|
||||
go askAgainSoon(ctx, inv)
|
||||
return
|
||||
case errors.As(err, &refusal):
|
||||
say("%v", err)
|
||||
@@ -122,3 +123,45 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.
|
||||
movingTheChain(ctx, inv)
|
||||
return nil
|
||||
}
|
||||
|
||||
// askAgain asks once more for each build request the move found never handed over (novox/hq issue 981 note 7 Q5): the
|
||||
// requests a merge could not ask for, which a person's hand `build` would otherwise have to repeat. **At most once**: a
|
||||
// request is claimed before it is asked and settled after, so a crash between loses the ask, which is where it stood.
|
||||
func askAgain(ctx context.Context, inv *inventory.Inventory) {
|
||||
pending, err := inv.RequestsToAskAgain(ctx)
|
||||
if err != nil {
|
||||
fmt.Printf("chain: the requests never handed over cannot be read: %v\n", err)
|
||||
return
|
||||
}
|
||||
for _, r := range pending {
|
||||
if claimed, err := inv.ClaimToAskAgain(ctx, r.ID); err != nil || !claimed {
|
||||
if err != nil {
|
||||
fmt.Printf("chain: %s cannot be claimed to be asked again: %v\n", r.ID, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
id, err := askABuild(ctx, buildSource{Repository: r.Repository, Seat: r.Seat}, r.Path, r.Ref)
|
||||
if err != nil {
|
||||
if serr := inv.AskedAgainAs(ctx, r.ID, "not asked again: "+err.Error()); serr != nil {
|
||||
fmt.Printf("chain: %s was not asked again, and that could not be kept: %v\n", r.ID, serr)
|
||||
}
|
||||
fmt.Printf("chain: %s was never handed over and cannot be asked again: %v\n", r.ID, err)
|
||||
continue
|
||||
}
|
||||
recordBuildRequest(ctx, inv, inventory.BuildRequest{ID: id, Repository: r.Repository, Seat: r.Seat, Path: r.Path,
|
||||
Ref: r.Ref, Commit: r.Commit, For: r.For})
|
||||
if err := inv.AskedAgainAs(ctx, r.ID, id); err != nil {
|
||||
fmt.Printf("chain: %s was asked again as %s, and that could not be kept: %v\n", r.ID, id, err)
|
||||
}
|
||||
fmt.Printf("chain: %s, never handed over, is asked again as %s\n", r.ID, id)
|
||||
}
|
||||
}
|
||||
|
||||
// askAgainSoon waits for the controller to be serving, which an ask needs, and then asks again.
|
||||
func askAgainSoon(ctx context.Context, inv *inventory.Inventory) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(time.Minute):
|
||||
askAgain(ctx, inv)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -259,3 +259,52 @@ func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T
|
||||
t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **The chain's buckets keep sixty-four revisions, the assignments' that were a mirror's included** (ADR 0408 §3 to §5,
|
||||
// issue 1089): a bucket made as a mirror — one revision, a marker age — is brought to the chain's options without the
|
||||
// server refusing the update, and a key written three times keeps three values.
|
||||
func TestTheChainBucketsKeepSixtyFourRevisionsOnTheBus(t *testing.T) {
|
||||
js, err := Dial(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer js.Close()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The bucket as the mirror made it.
|
||||
if _, err := api.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: AssignmentsBucket, History: 1,
|
||||
MaxValueSize: MirrorValueBytes, MaxBytes: MirrorBytes, Storage: jetstream.FileStorage,
|
||||
LimitMarkerTTL: MirrorMarkerTTL}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := js.EnsureControllerBuckets(); err != nil {
|
||||
t.Fatalf("a mirror's bucket is not brought to the chain's options: %v", err)
|
||||
}
|
||||
for _, name := range ChainBuckets() {
|
||||
kv, err := api.KeyValue(ctx, name)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
status, err := kv.Status(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg := status.(*jetstream.KeyValueBucketStatus).StreamInfo().Config
|
||||
if status.History() != ChainHistory || status.TTL() != 0 || cfg.MaxMsgSize < 0 || cfg.MaxMsgSize != ChainValueBytes {
|
||||
t.Errorf("%s keeps %d revisions, age %s, a value of %d bytes", name, status.History(), status.TTL(), cfg.MaxMsgSize)
|
||||
}
|
||||
for _, value := range []string{"one", "two", "three"} {
|
||||
if _, err := kv.Put(ctx, "k", []byte(value)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
history, err := kv.History(ctx, "k")
|
||||
if err != nil || len(history) != 3 {
|
||||
t.Errorf("%s keeps %d values of a key written three times: %v", name, len(history), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -53,9 +53,37 @@ type Source struct {
|
||||
}
|
||||
|
||||
// The states of a build record. A record in state asked is a request that no outcome has answered (issue 1091); one
|
||||
// without a state is an outcome.
|
||||
// without a state is an outcome, and keeps the Request it was asked with.
|
||||
const BuildAsked = "asked"
|
||||
|
||||
// The states of a request's Reask.
|
||||
const (
|
||||
ReaskPending = "pending"
|
||||
ReaskClaimed = "claimed"
|
||||
)
|
||||
|
||||
// Request is a build as it was asked for (novox/hq issue 325, ADR 0408 §1): kept in the build's own record under its id,
|
||||
// first alone as state asked and then beside the outcome that answered it.
|
||||
type Request struct {
|
||||
// Repository and Seat are the source as the mesh spells it; Path is the module's directory in it, Ref the branch or
|
||||
// commit asked, and Commit the commit that made the ask, where one did.
|
||||
Repository string `json:"repository,omitempty"`
|
||||
Seat string `json:"seat,omitempty"`
|
||||
Path string `json:"path,omitempty"`
|
||||
Ref string `json:"ref,omitempty"`
|
||||
Commit string `json:"commit,omitempty"`
|
||||
// For says who asked: merge, plan, build (a person), assign.
|
||||
For string `json:"for,omitempty"`
|
||||
// NotAsked is why the ask could not be handed over; OutcomeUnknown why an asker that handed it over stopped waiting.
|
||||
NotAsked string `json:"not-asked,omitempty"`
|
||||
OutcomeUnknown string `json:"outcome-unknown,omitempty"`
|
||||
At time.Time `json:"at,omitzero"`
|
||||
AtTerminal bool `json:"at-terminal,omitempty"`
|
||||
// Reask is, on a request the move found never handed over, "pending" until the controller asks it again, "claimed"
|
||||
// while it does, and the build id it was asked as once it has: published once, never twice (issue 981 note 7 Q5).
|
||||
Reask string `json:"reask,omitempty"`
|
||||
}
|
||||
|
||||
// Build is one attempt, whichever way it went, kept under its id. It names its module, its repository and commit, its
|
||||
// reuse key (novox/hq ADR 0350) and what it published (ADR 0408 §1).
|
||||
type Build struct {
|
||||
@@ -79,6 +107,7 @@ type Build struct {
|
||||
At time.Time `json:"at,omitzero"`
|
||||
ReuseKey string `json:"reuse-key,omitempty"`
|
||||
Published *Published `json:"published,omitempty"`
|
||||
Request *Request `json:"request,omitempty"`
|
||||
}
|
||||
|
||||
// Commit is one repository's commit a rollout carries.
|
||||
|
||||
@@ -88,17 +88,46 @@ func publishedOf(made []Artifact) *chain.Published {
|
||||
func chainBuildKey(id string) string { return id }
|
||||
|
||||
// chainRecordBuild keeps a build under its id. **Idempotent on the id**, as the table's insert was: a result arrives twice,
|
||||
// and the first record stands.
|
||||
// and the first record stands. A request nobody has answered is the record in state asked (issue 1091); the outcome
|
||||
// takes its place under the same key, and keeps the request it answers.
|
||||
func (i *Inventory) chainRecordBuild(ctx context.Context, c *chain.Chain, b Build) error {
|
||||
if b.At.IsZero() {
|
||||
b.At = time.Now().UTC()
|
||||
}
|
||||
if _, err := c.Builds.Create(ctx, chainBuildKey(b.ID), toChainBuild(b)); err != nil && !errors.Is(err, chain.ErrExists) {
|
||||
return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err)
|
||||
key := chainBuildKey(b.ID)
|
||||
for try := 0; try < casTries; try++ {
|
||||
next := toChainBuild(b)
|
||||
cur, found, err := c.Builds.Get(ctx, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err)
|
||||
}
|
||||
switch {
|
||||
case !found:
|
||||
_, err = c.Builds.Create(ctx, key, next)
|
||||
if errors.Is(err, chain.ErrExists) {
|
||||
continue
|
||||
}
|
||||
case cur.Value.State == chain.BuildAsked:
|
||||
next.Request = cur.Value.Request
|
||||
_, err = c.Builds.Update(ctx, key, next, cur.Revision)
|
||||
if errors.Is(err, chain.ErrMoved) {
|
||||
continue
|
||||
}
|
||||
default:
|
||||
// An outcome stands: the first record is the record.
|
||||
return i.RecordMirrored(ctx, b.ID, "", b.Mirrored)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err)
|
||||
}
|
||||
return i.RecordMirrored(ctx, b.ID, "", b.Mirrored)
|
||||
}
|
||||
return i.RecordMirrored(ctx, b.ID, "", b.Mirrored)
|
||||
return fmt.Errorf("build %s is not kept on the bus: it kept moving under the write", b.ID)
|
||||
}
|
||||
|
||||
// casTries is how often a write that lost a compare-and-set reads the key again and tries once more.
|
||||
const casTries = 8
|
||||
|
||||
// chainBuilds is every build of the bucket, whole, in no order.
|
||||
func chainBuilds(ctx context.Context, c *chain.Chain) ([]Build, error) {
|
||||
all, err := c.Builds.All(ctx)
|
||||
@@ -107,6 +136,10 @@ func chainBuilds(ctx context.Context, c *chain.Chain) ([]Build, error) {
|
||||
}
|
||||
out := make([]Build, 0, len(all))
|
||||
for _, k := range all {
|
||||
// A request nobody has answered is no build: it has no outcome to read.
|
||||
if k.Value.State == chain.BuildAsked {
|
||||
continue
|
||||
}
|
||||
out = append(out, fromChainBuild(k.Value))
|
||||
}
|
||||
return out, nil
|
||||
@@ -169,7 +202,8 @@ func (i *Inventory) chainBuildList(ctx context.Context, c *chain.Chain, module s
|
||||
|
||||
func (i *Inventory) chainBuildByID(ctx context.Context, c *chain.Chain, id string) (Build, bool, error) {
|
||||
k, ok, err := c.Builds.Get(ctx, chainBuildKey(id))
|
||||
if err != nil || !ok {
|
||||
if err != nil || !ok || k.Value.State == chain.BuildAsked {
|
||||
// A request nobody has answered has no outcome to read.
|
||||
return Build{}, false, err
|
||||
}
|
||||
b := fromChainBuild(k.Value)
|
||||
|
||||
@@ -36,6 +36,9 @@ type MoveReport struct {
|
||||
// Rows is how many records the tables held, Written how many keys this move made, and Present how many already
|
||||
// held a value (moved by an earlier start).
|
||||
Rows, Written, Present int
|
||||
// TableRows is how many rows of the tables those records came from: a request and its outcome are two rows and one
|
||||
// record.
|
||||
TableRows int
|
||||
// Largest is the biggest value, in bytes.
|
||||
Largest int
|
||||
// Already is a kind an earlier start moved: nothing was done.
|
||||
@@ -45,13 +48,13 @@ type MoveReport struct {
|
||||
// mover is one kind of the chain: its rows, read from its tables, as the keys and values they become.
|
||||
type mover struct {
|
||||
kind string
|
||||
// rows reads the tables: each record's key and its value, whole.
|
||||
rows func(ctx context.Context) ([]moveRow, error)
|
||||
// rows reads the tables: each record's key and its value, whole, and how many rows of the tables they came from.
|
||||
rows func(ctx context.Context) ([]moveRow, int, error)
|
||||
// create writes one record under its key; chain.ErrExists when the key holds a value.
|
||||
create func(ctx context.Context, c *chain.Chain, row moveRow) error
|
||||
// keys are the keys the bucket holds.
|
||||
keys func(ctx context.Context, c *chain.Chain) ([]string, error)
|
||||
// count is how many rows the tables hold now.
|
||||
// count is how many rows the tables hold now, counted as rows reports them.
|
||||
count func(ctx context.Context, tx pgx.Tx) (int, error)
|
||||
// lock keeps the tables from taking a row while the switch is recorded.
|
||||
lock string
|
||||
@@ -66,21 +69,46 @@ type moveRow struct {
|
||||
func (i *Inventory) movers() []mover {
|
||||
return []mover{{
|
||||
kind: KindBuilds,
|
||||
rows: func(ctx context.Context) ([]moveRow, error) {
|
||||
rows: func(ctx context.Context) ([]moveRow, int, error) {
|
||||
builds, err := i.allBuildRows(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, 0, err
|
||||
}
|
||||
out := make([]moveRow, 0, len(builds))
|
||||
requests, err := i.allRequestRows(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
// A build request is the build's record in state asked, and the request stays beside the outcome that
|
||||
// answered it: one record under one id (issue 1091).
|
||||
byID := make(map[string]*chain.Build, len(builds))
|
||||
var order []string
|
||||
for _, b := range builds {
|
||||
v := toChainBuild(b)
|
||||
body, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, moveRow{key: chainBuildKey(b.ID), value: v, size: len(body)})
|
||||
byID[b.ID] = &v
|
||||
order = append(order, b.ID)
|
||||
}
|
||||
return out, nil
|
||||
for _, r := range requests {
|
||||
if v, answered := byID[r.ID]; answered {
|
||||
v.Request = toRequest(r)
|
||||
continue
|
||||
}
|
||||
v := askedRecord(r)
|
||||
if r.NotAsked != "" {
|
||||
// Never handed over: asked again once, after the move (askAgain).
|
||||
v.Request.Reask = chain.ReaskPending
|
||||
}
|
||||
byID[r.ID] = &v
|
||||
order = append(order, r.ID)
|
||||
}
|
||||
out := make([]moveRow, 0, len(order))
|
||||
for _, id := range order {
|
||||
body, err := json.Marshal(byID[id])
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
out = append(out, moveRow{key: chainBuildKey(id), value: *byID[id], size: len(body)})
|
||||
}
|
||||
return out, len(builds) + len(requests), nil
|
||||
},
|
||||
create: func(ctx context.Context, c *chain.Chain, row moveRow) error {
|
||||
_, err := c.Builds.Create(ctx, row.key, row.value.(chain.Build))
|
||||
@@ -89,10 +117,10 @@ func (i *Inventory) movers() []mover {
|
||||
keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Builds.Keys(ctx) },
|
||||
count: func(ctx context.Context, tx pgx.Tx) (int, error) {
|
||||
var n int
|
||||
err := tx.QueryRow(ctx, `select count(*) from build`).Scan(&n)
|
||||
err := tx.QueryRow(ctx, `select (select count(*) from build) + (select count(*) from build_request)`).Scan(&n)
|
||||
return n, err
|
||||
},
|
||||
lock: `lock table build in share mode`,
|
||||
lock: `lock table build, build_request in share mode`,
|
||||
}}
|
||||
}
|
||||
|
||||
@@ -131,11 +159,11 @@ func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...a
|
||||
return report, nil
|
||||
}
|
||||
for pass := 1; pass <= movePasses; pass++ {
|
||||
rows, err := m.rows(ctx)
|
||||
rows, tableRows, err := m.rows(ctx)
|
||||
if err != nil {
|
||||
return report, fmt.Errorf("the %s cannot be read from the tables: %w", m.kind, err)
|
||||
}
|
||||
report.Rows, report.Largest = len(rows), 0
|
||||
report.Rows, report.TableRows, report.Largest = len(rows), tableRows, 0
|
||||
// Every value measured before any is written: a value over the bus's largest stops the move with nothing
|
||||
// written (ADR 0408 §6), never cut.
|
||||
for _, r := range rows {
|
||||
@@ -152,6 +180,8 @@ func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...a
|
||||
report.Written++
|
||||
case errors.Is(err, chain.ErrExists):
|
||||
report.Present++
|
||||
case errors.Is(err, chain.ErrBadKey):
|
||||
return report, &Refusal{m.kind, fmt.Sprintf("%s cannot be a key of the bus: %v", r.key, err)}
|
||||
default:
|
||||
return report, fmt.Errorf("%s %s is not written to the bus: %w", m.kind, r.key, err)
|
||||
}
|
||||
@@ -199,7 +229,7 @@ func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if n != r.Rows {
|
||||
if n != r.TableRows {
|
||||
return false, nil
|
||||
}
|
||||
counts, _ := json.Marshal(map[string]int{m.kind: r.Rows})
|
||||
@@ -263,3 +293,25 @@ func (i *Inventory) allBuildRows(ctx context.Context) ([]Build, error) {
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// allRequestRows is every build request of the table.
|
||||
func (i *Inventory) allRequestRows(ctx context.Context) ([]BuildRequest, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select id, repository, seat, source_path, ref, commit_hash, asked_for, coalesce(not_asked, ''),
|
||||
coalesce(outcome_unknown, ''), asked_at, at_terminal
|
||||
from build_request order by asked_at, id`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []BuildRequest
|
||||
for rows.Next() {
|
||||
var a BuildRequest
|
||||
if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked,
|
||||
&a.OutcomeUnknown, &a.At, &a.AtTerminal); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
@@ -6,7 +6,9 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/chain"
|
||||
)
|
||||
@@ -270,3 +272,153 @@ func TestBuildsOnTheBusAreReadBackByEveryReader(t *testing.T) {
|
||||
t.Fatalf("fingerprints %v: %v", prints, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A build request row as the store held it before the move.
|
||||
func aRequestRow(t *testing.T, inv *Inventory, id, notAsked string) {
|
||||
t.Helper()
|
||||
inv.ExecForTest(t,
|
||||
`insert into build_request (id, repository, seat, source_path, ref, commit_hash, asked_for, not_asked, asked_at)
|
||||
values ($1::text, 'novox/notes', '', 'modules/notes', 'main', 'c1', 'merge', nullif($2::text, ''), now() - interval '1 hour')`,
|
||||
id, notAsked)
|
||||
}
|
||||
|
||||
// **A request is the build record in state asked, and its outcome takes its place and keeps it** (ADR 0408 §1, issue
|
||||
// 1091): no outcome to read until it comes, the marks of a request that nobody answered kept in the record, and none
|
||||
// after the outcome is heard. The table of requests takes none.
|
||||
func TestARequestIsTheBuildInStateAskedAndItsOutcomeTakesItsPlace(t *testing.T) {
|
||||
inv := fresh(t)
|
||||
ctx := context.Background()
|
||||
ask := BuildRequest{ID: "build-1", Repository: "novox/notes", Path: "modules/notes", Ref: "main", For: "build", AtTerminal: true}
|
||||
if err := inv.RecordBuildRequest(ctx, ask); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordBuildRequest(ctx, ask); err != nil {
|
||||
t.Fatalf("a request kept twice: %v", err)
|
||||
}
|
||||
if n := rowsInTheTable(t, inv, "build_request"); n != "0" {
|
||||
t.Fatalf("the table took %s requests", n)
|
||||
}
|
||||
if builds, err := inv.Builds(ctx, "", 10); err != nil || len(builds) != 0 {
|
||||
t.Fatalf("a request nobody answered is a build: %+v %v", builds, err)
|
||||
}
|
||||
if _, found, _ := inv.BuildByID(ctx, "build-1"); found {
|
||||
t.Fatal("a request nobody answered has an outcome")
|
||||
}
|
||||
got, found, err := inv.BuildRequestByID(ctx, "build-1")
|
||||
if err != nil || !found || got.Repository != "novox/notes" || got.For != "build" || !got.AtTerminal || got.At.IsZero() {
|
||||
t.Fatalf("the request reads back as %+v %v %v", got, found, err)
|
||||
}
|
||||
if err := inv.MarkOutcomeUnknown(ctx, "build-1", "stopped waiting"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
asked, err := inv.RequestsNamed(ctx, "notes")
|
||||
if err != nil || len(asked) != 1 || asked[0].Heard || asked[0].OutcomeUnknown != "stopped waiting" {
|
||||
t.Fatalf("the request is %+v: %v", asked, err)
|
||||
}
|
||||
if names, err := inv.RequestedNames(ctx); err != nil || !reflect.DeepEqual(names, []string{"notes"}) {
|
||||
t.Fatalf("requested names %v: %v", names, err)
|
||||
}
|
||||
|
||||
// The outcome comes: the same key, now a build, with the request beside it.
|
||||
built := aBuild("build-1", "notes", "")
|
||||
if err := inv.RecordBuild(ctx, built); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if one, found, err := inv.BuildByID(ctx, "build-1"); err != nil || !found || one.Module != "notes" {
|
||||
t.Fatalf("the outcome reads as %+v %v %v", one, found, err)
|
||||
}
|
||||
answered, err := inv.RequestsNamed(ctx, "notes")
|
||||
if err != nil || len(answered) != 1 || !answered[0].Heard || answered[0].Module != "notes" || answered[0].Failed != "" {
|
||||
t.Fatalf("the request answered is %+v: %v", answered, err)
|
||||
}
|
||||
if again, _, _ := inv.BuildRequestByID(ctx, "build-1"); again.For != "build" || again.OutcomeUnknown != "stopped waiting" {
|
||||
t.Fatalf("the request was lost with the outcome: %+v", again)
|
||||
}
|
||||
// Heard: the marks of an unanswered request change nothing.
|
||||
if err := inv.MarkNotAsked(ctx, "build-1", "late"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again, _, _ := inv.BuildRequestByID(ctx, "build-1"); again.NotAsked != "" {
|
||||
t.Fatalf("an answered request was marked not asked: %+v", again)
|
||||
}
|
||||
if history, _ := inv.chain.Builds.History(ctx, "build-1"); len(history) != 3 {
|
||||
// asked, marked unknown, answered.
|
||||
t.Fatalf("the record has %d values", len(history))
|
||||
}
|
||||
// An outcome heard before its request is kept, and the request lands beside it.
|
||||
if err := inv.RecordBuild(ctx, aBuild("build-2", "wiki", "")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordBuildRequest(ctx, BuildRequest{ID: "build-2", Repository: "novox/wiki", Path: "modules/wiki", For: "plan"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if late, found, _ := inv.BuildRequestByID(ctx, "build-2"); !found || late.For != "plan" {
|
||||
t.Fatalf("the request kept after its outcome reads %+v", late)
|
||||
}
|
||||
// A request older than the time they are kept is no longer listed.
|
||||
if err := inv.RecordBuildRequest(ctx, BuildRequest{ID: "build-old", Repository: "novox/old", Path: "modules/old",
|
||||
At: time.Now().Add(-2 * KeptFor)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if names, _ := inv.RequestedNames(ctx); slices.Contains(names, "old") {
|
||||
t.Fatalf("a request past its time is listed: %v", names)
|
||||
}
|
||||
}
|
||||
|
||||
// **The move carries the build requests**: an answered one beside its outcome, an unanswered one as state asked, and a
|
||||
// request never handed over as state asked marked to be asked again once (issue 981 note 7 Q5), which is claimed before it
|
||||
// is asked and so asked at most once.
|
||||
func TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain(t *testing.T) {
|
||||
inv, finish := ForTestBefore(t, 105)
|
||||
finish()
|
||||
ctx := context.Background()
|
||||
aBuildRow(t, inv, "build-1", "notes", "c1", "")
|
||||
aRequestRow(t, inv, "build-1", "")
|
||||
aRequestRow(t, inv, "build-2", "")
|
||||
aRequestRow(t, inv, "not-asked-c1-modules-notes", "no seat holder")
|
||||
|
||||
if requests, err := inv.RequestsNamed(ctx, "notes"); err != nil || len(requests) != 3 {
|
||||
t.Fatalf("before the move the table holds %d requests: %v", len(requests), err)
|
||||
}
|
||||
reports, err := inv.MoveChain(ctx, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if reports[0].Rows != 3 || reports[0].TableRows != 4 {
|
||||
t.Fatalf("the move reported %+v", reports[0])
|
||||
}
|
||||
one, _, _ := inv.chain.Builds.Get(ctx, "build-1")
|
||||
if one.Value.State != "" || one.Value.Request == nil || one.Value.Request.For != "merge" || one.Value.Module != "notes" {
|
||||
t.Fatalf("an answered request moved as %+v", one.Value)
|
||||
}
|
||||
two, _, _ := inv.chain.Builds.Get(ctx, "build-2")
|
||||
if two.Value.State != chain.BuildAsked || two.Value.Request.Reask != "" {
|
||||
t.Fatalf("an unanswered request moved as %+v", two.Value)
|
||||
}
|
||||
never, _, _ := inv.chain.Builds.Get(ctx, "not-asked-c1-modules-notes")
|
||||
if never.Value.State != chain.BuildAsked || never.Value.Request.NotAsked != "no seat holder" || never.Value.Request.Reask != chain.ReaskPending {
|
||||
t.Fatalf("a request never handed over moved as %+v", never.Value)
|
||||
}
|
||||
requests, err := inv.RequestsNamed(ctx, "notes")
|
||||
if err != nil || len(requests) != 3 {
|
||||
t.Fatalf("after the move the bucket holds %d requests: %v", len(requests), err)
|
||||
}
|
||||
toAsk, err := inv.RequestsToAskAgain(ctx)
|
||||
if err != nil || len(toAsk) != 1 || toAsk[0].ID != "not-asked-c1-modules-notes" {
|
||||
t.Fatalf("to ask again: %+v %v", toAsk, err)
|
||||
}
|
||||
first, err := inv.ClaimToAskAgain(ctx, toAsk[0].ID)
|
||||
second, err2 := inv.ClaimToAskAgain(ctx, toAsk[0].ID)
|
||||
if err != nil || err2 != nil || !first || second {
|
||||
t.Fatalf("claimed %v then %v: %v %v", first, second, err, err2)
|
||||
}
|
||||
if left, _ := inv.RequestsToAskAgain(ctx); len(left) != 0 {
|
||||
t.Fatalf("a claimed request is still to be asked: %+v", left)
|
||||
}
|
||||
if err := inv.AskedAgainAs(ctx, toAsk[0].ID, "build-9"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if done, _, _ := inv.BuildRequestByID(ctx, toAsk[0].ID); done.NotAsked != "no seat holder" {
|
||||
t.Fatalf("the request lost its words: %+v", done)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
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, on, err := i.ChainReads(ctx, KindBuilds)
|
||||
if err != nil || !on {
|
||||
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, on, err := i.ChainReads(ctx, KindBuilds)
|
||||
if err != nil || !on {
|
||||
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, 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 })
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/chain"
|
||||
)
|
||||
|
||||
// **A request is the build record in state asked, under the id its outcome will carry** (ADR 0408 §1, issue 1091): the
|
||||
// whole ask is in it, with a path trimmed as the table kept it.
|
||||
func TestABuildRequestIsTheBuildRecordInStateAsked(t *testing.T) {
|
||||
at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC)
|
||||
ask := BuildRequest{ID: "build-1", Repository: "https://forge.invalid/notes.git", Seat: "", Path: "/modules/notes/", Ref: "main",
|
||||
Commit: "c1", For: "merge", NotAsked: "", OutcomeUnknown: "", At: at, AtTerminal: true}
|
||||
record := askedRecord(ask)
|
||||
if record.State != chain.BuildAsked || record.ID != "build-1" || record.Path != "modules/notes" || !record.Asked.Equal(at) {
|
||||
t.Fatalf("the request is %+v", record)
|
||||
}
|
||||
if got := fromRequest(record.ID, *record.Request); !reflect.DeepEqual(got, BuildRequest{ID: "build-1",
|
||||
Repository: ask.Repository, Path: "modules/notes", Ref: "main", Commit: "c1", For: "merge", At: at, AtTerminal: true}) {
|
||||
t.Fatalf("the request reads back as %+v", got)
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,6 @@ package inventory
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -70,6 +69,11 @@ type RequestOutcome struct {
|
||||
// RecordBuildRequest keeps one build request, once it is asked. Idempotent on the id. Requests older than
|
||||
// KeptFor are deleted in the same act.
|
||||
func (i *Inventory) RecordBuildRequest(ctx context.Context, a BuildRequest) error {
|
||||
if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil {
|
||||
return err
|
||||
} else if on {
|
||||
return i.chainRecordBuildRequest(ctx, c, a)
|
||||
}
|
||||
at := a.At
|
||||
if at.IsZero() {
|
||||
at = time.Now().UTC()
|
||||
@@ -93,6 +97,11 @@ func (i *Inventory) RecordBuildRequest(ctx context.Context, a BuildRequest) erro
|
||||
|
||||
// BuildRequestByID is the build request kept under this id, and whether one was kept.
|
||||
func (i *Inventory) BuildRequestByID(ctx context.Context, id string) (BuildRequest, bool, error) {
|
||||
if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil {
|
||||
return BuildRequest{}, false, err
|
||||
} else if on {
|
||||
return i.chainBuildRequestByID(ctx, c, id)
|
||||
}
|
||||
var a BuildRequest
|
||||
err := i.store.Pool().QueryRow(ctx,
|
||||
`select id, repository, seat, source_path, ref, commit_hash, asked_for, at_terminal, asked_at
|
||||
@@ -107,6 +116,11 @@ func (i *Inventory) BuildRequestByID(ctx context.Context, id string) (BuildReque
|
||||
// MarkNotAsked says a kept build request was never handed over: the words are kept, unless its outcome was
|
||||
// heard first.
|
||||
func (i *Inventory) MarkNotAsked(ctx context.Context, id, why string) error {
|
||||
if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil {
|
||||
return err
|
||||
} else if on {
|
||||
return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.NotAsked = why })
|
||||
}
|
||||
if heard, err := i.outcomeHeard(ctx, id); err != nil || heard {
|
||||
return err
|
||||
}
|
||||
@@ -117,6 +131,11 @@ func (i *Inventory) MarkNotAsked(ctx context.Context, id, why string) error {
|
||||
// MarkOutcomeUnknown says the asker of a build it handed over stopped waiting for it: asked, outcome unknown.
|
||||
// The build may still run; its outcome, when heard, is the last word.
|
||||
func (i *Inventory) MarkOutcomeUnknown(ctx context.Context, id, why string) error {
|
||||
if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil {
|
||||
return err
|
||||
} else if on {
|
||||
return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.OutcomeUnknown = why })
|
||||
}
|
||||
if heard, err := i.outcomeHeard(ctx, id); err != nil || heard {
|
||||
return err
|
||||
}
|
||||
@@ -124,14 +143,8 @@ func (i *Inventory) MarkOutcomeUnknown(ctx context.Context, id, why string) erro
|
||||
return err
|
||||
}
|
||||
|
||||
// outcomeHeard says a build of this id was recorded: its outcome was taken in.
|
||||
// outcomeHeard says a build of this id was recorded in the table: its outcome was taken in.
|
||||
func (i *Inventory) outcomeHeard(ctx context.Context, id string) (bool, error) {
|
||||
if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil {
|
||||
return false, err
|
||||
} else if on {
|
||||
_, heard, err := c.Builds.Get(ctx, chainBuildKey(id))
|
||||
return heard, err
|
||||
}
|
||||
var heard bool
|
||||
err := i.store.Pool().QueryRow(ctx, `select exists (select 1 from build where id = $1)`, id).Scan(&heard)
|
||||
return heard, err
|
||||
@@ -199,45 +212,6 @@ func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) {
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// chainRequests is requests with each outcome read from the builds bucket, where the table's join read it.
|
||||
func (i *Inventory) chainRequests(ctx context.Context, c *chain.Chain) ([]RequestOutcome, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select id, repository, seat, source_path, ref, commit_hash, asked_for,
|
||||
coalesce(not_asked, ''), coalesce(outcome_unknown, ''), asked_at
|
||||
from build_request
|
||||
order by asked_at desc, id desc`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []RequestOutcome
|
||||
for rows.Next() {
|
||||
var a RequestOutcome
|
||||
if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked,
|
||||
&a.OutcomeUnknown, &a.At); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a.HeardAt = a.At
|
||||
out = append(out, a)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
built, err := c.Builds.All(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err)
|
||||
}
|
||||
for k := range out {
|
||||
if b, heard := built[chainBuildKey(out[k].ID)]; heard {
|
||||
out[k].Heard, out[k].Failed, out[k].Module, out[k].HeardAt = true, b.Value.Failed, b.Value.Module, b.Value.At
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// The states of a pending assignment.
|
||||
const (
|
||||
PendingWaiting = "waiting"
|
||||
|
||||
Vendored
+2
@@ -145,3 +145,5 @@
|
||||
./internal/inventory TestTheMoveRefusesOnACountThatDiffers
|
||||
./internal/inventory TestTheMoveWithoutTheLeaseWritesNothing
|
||||
./internal/inventory TestBuildsOnTheBusAreReadBackByEveryReader
|
||||
./internal/inventory TestARequestIsTheBuildInStateAskedAndItsOutcomeTakesItsPlace
|
||||
./internal/inventory TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain
|
||||
|
||||
Reference in New Issue
Block a user