Give each pending assignment its own condition, and keep a raised row until it clears
mesh/merge-gate pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus step; every machine composes with the change as it…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery-group group fix/assign-says-why-a-module-is-not-there delivering: 1 of 2 delivered
mesh/delivery superseded: a newer delivery to the same trunk took over its walk
mesh/merge-gate pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus step; every machine composes with the change as it…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery-group group fix/assign-says-why-a-module-is-not-there delivering: 1 of 2 delivered
mesh/delivery superseded: a newer delivery to the same trunk took over its walk
Second review of #150: one key per machine and module let a newer failure be cleared in the tick that raised it, pruning could orphan an open condition, and a build no longer waited for read as never asked although it may still run.
This commit is contained in:
@@ -68,3 +68,13 @@ func ForTest(t *testing.T) *Inventory {
|
||||
}
|
||||
return inv
|
||||
}
|
||||
|
||||
// RemoveNodeForTest removes a machine's record as a person removing it from the store would: no verb of the
|
||||
// mesh removes one yet, and what hangs off it must go with it.
|
||||
func (i *Inventory) RemoveNodeForTest(ctx context.Context, name string) (bool, error) {
|
||||
tag, err := i.store.Pool().Exec(ctx, `delete from node where name = $1`, name)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return tag.RowsAffected() == 1, nil
|
||||
}
|
||||
|
||||
@@ -8,8 +8,10 @@
|
||||
|
||||
-- Every build the controller asked for: a merge's new module, a plan's tier, a person's `build`, an `assign`
|
||||
-- with build. Kept once the ask is made, never before. Its outcome is the `build` row of the same id;
|
||||
-- an ask without one is still running, or was lost. not_asked is why the ask could not be made or waited
|
||||
-- for; for a merge that could not ask, the id is the controller's and no build's.
|
||||
-- an ask without one is still running, or was lost. not_asked is why the ask could not be handed over; for a
|
||||
-- merge that could not ask, the id is the controller's and no build's. outcome_unknown is why an asker that
|
||||
-- handed it over stopped waiting: the build may still run, and is read as in flight until its outcome or
|
||||
-- its bound.
|
||||
create table build_request (
|
||||
id text primary key,
|
||||
repository text not null,
|
||||
@@ -19,6 +21,7 @@ create table build_request (
|
||||
commit_hash text not null default '',
|
||||
asked_for text not null default '',
|
||||
not_asked text,
|
||||
outcome_unknown text,
|
||||
asked_at timestamptz not null default now()
|
||||
);
|
||||
create index build_request_asked_at on build_request (asked_at);
|
||||
@@ -28,7 +31,7 @@ create index build_request_asked_at on build_request (asked_at);
|
||||
-- assignment was refused then), expired (the build failed, was not registered, or said nothing within its
|
||||
-- bound) or withdrawn (`unassign`). raised_at is when an expiry or refusal was raised as a condition,
|
||||
-- acknowledged_at when a person took it back with `unassign`, cleared_at when its condition was cleared.
|
||||
-- Gone with its machine. Ended rows are deleted after 30 days.
|
||||
-- Gone with its machine. Ended rows are deleted after 30 days, once any condition raised for them is cleared.
|
||||
create table pending_assignment (
|
||||
id bigserial primary key,
|
||||
node uuid not null references node (id) on delete cascade,
|
||||
|
||||
@@ -32,9 +32,12 @@ type BuildRequest struct {
|
||||
Commit string
|
||||
// For says who asked: "merge", "plan", "build" (a person), "assign".
|
||||
For string
|
||||
// NotAsked is why the ask could not be made, or not waited for; empty when it was.
|
||||
// NotAsked is why the ask could not be handed over; empty when it was.
|
||||
NotAsked string
|
||||
At time.Time
|
||||
// OutcomeUnknown is why an asker that handed the build over stopped waiting for it: asked, outcome
|
||||
// unknown. Read as in flight until its outcome or its bound.
|
||||
OutcomeUnknown string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
// Name is the module this request is expected to register, read from its directory: the last element of
|
||||
@@ -80,8 +83,8 @@ func (i *Inventory) RecordBuildRequest(ctx context.Context, a BuildRequest) erro
|
||||
return err
|
||||
}
|
||||
|
||||
// MarkNotAsked says a build request was asked and its asker could not hand it over, or stopped waiting for
|
||||
// it: the words are kept, unless its outcome was heard first.
|
||||
// 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 {
|
||||
_, err := i.store.Pool().Exec(ctx,
|
||||
`update build_request set not_asked = $2
|
||||
@@ -89,6 +92,15 @@ func (i *Inventory) MarkNotAsked(ctx context.Context, id, why string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// 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 {
|
||||
_, err := i.store.Pool().Exec(ctx,
|
||||
`update build_request set outcome_unknown = $2
|
||||
where id = $1 and not exists (select 1 from build where build.id = $1)`, id, why)
|
||||
return err
|
||||
}
|
||||
|
||||
// RequestsNamed is every build request kept whose directory names the module, or whose build said it built
|
||||
// it, newest first, each with its outcome.
|
||||
func (i *Inventory) RequestsNamed(ctx context.Context, name string) ([]RequestOutcome, error) {
|
||||
@@ -126,7 +138,7 @@ func (i *Inventory) RequestedNames(ctx context.Context) ([]string, error) {
|
||||
func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select a.id, a.repository, a.seat, a.source_path, a.ref, a.commit_hash, a.asked_for,
|
||||
coalesce(a.not_asked, ''), a.asked_at,
|
||||
coalesce(a.not_asked, ''), coalesce(a.outcome_unknown, ''), a.asked_at,
|
||||
b.id is not null, coalesce(b.failed, ''), coalesce(b.module, ''), coalesce(b.at, a.asked_at)
|
||||
from build_request a left join build b on b.id = a.id
|
||||
order by a.asked_at desc, a.id desc`)
|
||||
@@ -138,7 +150,7 @@ func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) {
|
||||
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.At, &a.Heard, &a.Failed, &a.Module, &a.HeardAt); err != nil {
|
||||
&a.OutcomeUnknown, &a.At, &a.Heard, &a.Failed, &a.Module, &a.HeardAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, a)
|
||||
@@ -342,13 +354,34 @@ func (i *Inventory) SettlePending(ctx context.Context, id int64, from, state, no
|
||||
return tag.RowsAffected() == 1, nil
|
||||
}
|
||||
|
||||
// ForgetEndedPending deletes ended pending assignments settled more than KeptFor ago, and says how many.
|
||||
// ForgetEndedPending deletes ended pending assignments settled more than KeptFor ago, and says how many. One
|
||||
// raised as a condition is kept until that condition is cleared: deleting it would leave the condition
|
||||
// with nothing to answer it.
|
||||
func (i *Inventory) ForgetEndedPending(ctx context.Context, now time.Time) (int64, error) {
|
||||
tag, err := i.store.Pool().Exec(ctx,
|
||||
`delete from pending_assignment where state not in ('waiting', 'applying') and settled_at < $1`,
|
||||
`delete from pending_assignment where state not in ('waiting', 'applying') and settled_at < $1
|
||||
and (raised_at is null or cleared_at is not null)`,
|
||||
now.Add(-KeptFor))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return tag.RowsAffected(), nil
|
||||
}
|
||||
|
||||
// PendingKnown says which of these pending assignments are still on record.
|
||||
func (i *Inventory) PendingKnown(ctx context.Context, ids []int64) (map[int64]bool, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `select id from pending_assignment where id = any($1)`, ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[int64]bool{}
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[id] = true
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
@@ -76,6 +76,11 @@ func (b *natsBuilds) Ask(ctx context.Context, request BuildRequest) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ErrNotHandedOver is a Submit that failed before the build was handed to the build seat: nothing runs. Any
|
||||
// other error from Submit came after: the build may still run, and its outcome may still come (novox/hq
|
||||
// issue 325).
|
||||
var ErrNotHandedOver = errors.New("the build was not handed over")
|
||||
|
||||
func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest,
|
||||
wait time.Duration) (BuildResult, error) {
|
||||
|
||||
@@ -84,23 +89,23 @@ func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest,
|
||||
// the same event on EVENTS, which the controller records.
|
||||
outcomes, err := b.js.Conn().SubscribeSync(BuildOutcomeOf(b.role()))
|
||||
if err != nil {
|
||||
return BuildResult{}, fmt.Errorf("cannot listen for a build's outcome: %w", err)
|
||||
return BuildResult{}, fmt.Errorf("%w: cannot listen for a build's outcome: %w", ErrNotHandedOver, err)
|
||||
}
|
||||
defer func() { _ = outcomes.Unsubscribe() }()
|
||||
if err := b.js.Conn().Flush(); err != nil {
|
||||
return BuildResult{}, err
|
||||
return BuildResult{}, fmt.Errorf("%w: %w", ErrNotHandedOver, err)
|
||||
}
|
||||
|
||||
body, err := json.Marshal(request)
|
||||
if err != nil {
|
||||
return BuildResult{}, err
|
||||
return BuildResult{}, fmt.Errorf("%w: %w", ErrNotHandedOver, err)
|
||||
}
|
||||
// Into the role's work queue and awaited: work the bus never accepted must fail here rather than
|
||||
// be assumed, because nothing else will ever say so.
|
||||
publish, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil {
|
||||
return BuildResult{}, fmt.Errorf("cannot submit a build: %w", err)
|
||||
return BuildResult{}, fmt.Errorf("%w: cannot submit a build: %w", ErrNotHandedOver, err)
|
||||
}
|
||||
|
||||
waiting, cancelWait := context.WithTimeout(ctx, wait)
|
||||
|
||||
Reference in New Issue
Block a user