Assemble merges in a rolling window and walk each batch once (hq ADR 0276, issue 362)

Every merge opened a walk and the next merge of the branch superseded it: two
catalogue merges 18 s apart left a walk no delivery held, and the operator
started it by hand 58 minutes later. A merge now joins the open batch, kept in
the store (migration 0089), which is cut into one walk when no merge came for
merge-window (90 s) or at merge-window-at-most (10 min): one commit per
repository, the latest of its branch, with every file the batch's merges
changed. One walk at a time; a started walk is never superseded, a waiting one
is folded into the next. The walk names every merge it answers on the wire
(delivery.merges, taken_over_by, batch). A failed walk walks its earlier merges
alone, newest first, until one is delivered. A delivery group's order becomes
tier edges inside the walk. plans shows the batch assembling; S18 and S19
bound its waits; S16 names the merges a waiting walk answers.

internal/inventory/batches.go

@@ -0,0 +1,117 @@
+package inventory
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// The merges a batch or walk answers (novox/hq ADR 0276): every merge the controller heard into a branch a
+// module follows is kept here once, with the batch or walk that answers it. A merge is never dropped and never
+// walked twice: its key is its repository and commit, and it belongs to one plan at a time.
+
+// BatchedMerge is one merge as the controller heard it.
+type BatchedMerge struct {
+	Repository string
+	Branch     string
+	Commit     string
+	// Merged is when the forge made it: the branch's order. Heard is when the controller heard it: what the
+	// window is measured from.
+	Merged time.Time
+	Heard  time.Time
+	// Event is the forge's announcement, what the cut plans from.
+	Event json.RawMessage
+	// Plan is the batch or walk answering it; empty while it waits to be walked alone.
+	Plan string
+	// Alone says it is walked on its own commit, after a failed walk that carried it in a later one.
+	Alone bool
+}
+
+// AddMerge keeps a merge for a plan; false when the merge was kept already, for whatever plan.
+func (i *Inventory) AddMerge(ctx context.Context, m BatchedMerge) (bool, error) {
+	tag, err := i.store.Pool().Exec(ctx,
+		`insert into batched_merge (repository, branch, commit_hash, merged_at, heard_at, event, plan_id, alone)
+		 values (lower($1), $2, $3, $4, $5, $6, nullif($7, ''), $8) on conflict do nothing`,
+		m.Repository, m.Branch, m.Commit, mergedAt(m.Merged), m.Heard, []byte(m.Event), m.Plan, m.Alone)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// MergeOf is one kept merge; false when it was never heard.
+func (i *Inventory) MergeOf(ctx context.Context, repository, commit string) (BatchedMerge, bool, error) {
+	ms, err := i.merges(ctx, `where repository = lower($1) and commit_hash = $2`, repository, commit)
+	if err != nil || len(ms) == 0 {
+		return BatchedMerge{}, false, err
+	}
+	return ms[0], true, nil
+}
+
+// MergesOf is every merge a plan answers, oldest merge first.
+func (i *Inventory) MergesOf(ctx context.Context, plan string) ([]BatchedMerge, error) {
+	return i.merges(ctx, `where plan_id = $1 order by merged_at nulls last, heard_at`, plan)
+}
+
+// LaterMergesOf is every merge of a repository's branch made after a moment and answered by a plan, the
+// newest first.
+func (i *Inventory) LaterMergesOf(ctx context.Context, repository, branch string, after time.Time) ([]BatchedMerge, error) {
+	return i.merges(ctx, `where repository = lower($1) and branch = $2 and merged_at > $3 and plan_id is not null
+		order by merged_at desc`, repository, branch, after)
+}
+
+// AloneMerges is every merge waiting to be walked on its own commit, the newest first (ADR 0276 decision 3).
+func (i *Inventory) AloneMerges(ctx context.Context) ([]BatchedMerge, error) {
+	return i.merges(ctx, `where plan_id is null and alone order by merged_at desc nulls last, heard_at desc`)
+}
+
+// AnswerMerge moves a merge to the plan that answers it now; an empty plan with alone leaves it waiting to
+// be walked on its own commit.
+func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan string, alone bool) error {
+	_, err := i.store.Pool().Exec(ctx,
+		`update batched_merge set plan_id = nullif($3, ''), alone = $4 where repository = lower($1) and commit_hash = $2`,
+		repository, commit, plan, alone)
+	return err
+}
+
+// Batches is every batch not yet cut: assembling or queued, oldest first. One at a time is the rule; more
+// are read so a fault is seen.
+func (i *Inventory) Batches(ctx context.Context) ([]Plan, error) {
+	return i.plans(ctx, `where state in ('assembling', 'queued') order by created`)
+}
+
+func (i *Inventory) merges(ctx context.Context, tail string, args ...any) ([]BatchedMerge, error) {
+	rows, err := i.store.Pool().Query(ctx,
+		`select repository, branch, commit_hash, merged_at, heard_at, event, coalesce(plan_id, ''), alone
+		   from batched_merge `+tail, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []BatchedMerge
+	for rows.Next() {
+		var m BatchedMerge
+		var merged *time.Time
+		var event []byte
+		if err := rows.Scan(&m.Repository, &m.Branch, &m.Commit, &merged, &m.Heard, &event, &m.Plan, &m.Alone); err != nil {
+			return nil, err
+		}
+		if merged != nil {
+			m.Merged = merged.UTC()
+		}
+		m.Heard = m.Heard.UTC()
+		m.Event = event
+		out = append(out, m)
+	}
+	if errors.Is(rows.Err(), pgx.ErrNoRows) {
+		return nil, nil
+	}
+	return out, rows.Err()
+}
+
+// SameRepository says two spellings of a repository are one.
+func SameRepository(a, b string) bool { return strings.EqualFold(a, b) }

internal/inventory/durations.go

@@ -125,6 +125,9 @@ func planTierLeft(ctx context.Context, tx pgx.Tx, p Plan, now time.Time) (entere
 	if err != nil {
 		return time.Time{}, err
 	}
+	if oldState == PlanAssembling || oldState == PlanQueued {
+		return now, nil // a batch cut into a walk enters its first tier now (novox/hq ADR 0276)
+	}
 	wasOpen := oldState == PlanBuilding || oldState == PlanRolling
 	if !wasOpen || (oldTier == p.Tier && p.Open()) {
 		return since, nil // still in the tier, or already ended

internal/inventory/migrations/0089-merges-are-assembled-in-a-rolling-window.sql

@@ -0,0 +1,28 @@
+-- Merges are assembled in a rolling window, and each batch is delivered by one walk (novox/hq ADR 0276,
+-- issue 362).
+--
+-- Every merge opened a walk of its own, and the next merge of the branch superseded it: two catalogue merges
+-- eighteen seconds apart on 2026-10-10 left a walk nobody gave its word, and the operator started it by hand
+-- 58 minutes later. A merge now joins the open batch, which is cut into one walk when no merge came for the
+-- window's length, with at most one commit per repository.
+--
+-- A walk names every repository's commit it carries (commits), beside the one its first columns keep for a
+-- reader that knows only one. Null for a walk kept before this: its repository and commit are the whole of it.
+alter table release_plan add column commits jsonb;
+
+-- Every merge the controller heard into a branch a module follows, and the batch or walk that answers it:
+-- what the window is measured from (heard_at), what the cut plans from (event, the forge's announcement), and
+-- what keeps a restarted controller from losing or doubling a merge (the key). plan_id is null for an
+-- earlier merge a failed walk carried, which waits to be walked on its own commit (alone).
+create table batched_merge (
+    repository  text        not null,
+    branch      text        not null,
+    commit_hash text        not null,
+    merged_at   timestamptz,
+    heard_at    timestamptz not null,
+    event       jsonb       not null,
+    plan_id     text,
+    alone       boolean     not null default false,
+    primary key (repository, commit_hash)
+);
+create index batched_merge_plan on batched_merge (plan_id);

internal/inventory/plans.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -45,11 +46,77 @@ type Plan struct {
 	// Release is set on a release plan (novox/hq ADR 0236): not a merge's, but the builds waiting for a
 	// gate, walked through the machines one at a time.
 	Release *PlanRelease `json:"release,omitempty"`
-	// Delivery is set on a walk that waits for its delivery's word (novox/hq ADR 0239): nil for a walk on
-	// the controller's own path, which starts at the merge as every plan did before.
+	// Delivery is set on a walk that waits for its delivery's word (novox/hq ADR 0239), and on every walk and
+	// batch that names the merges it answers (ADR 0276): Awaits empty for a walk on the controller's own path,
+	// which starts when it is cut.
 	Delivery *PlanDelivery `json:"delivery,omitempty"`
+	// Commits is every repository's commit the walk carries, one per repository (novox/hq ADR 0276): a batch's
+	// walk can carry several. Repository and Commit above keep one of them, for a reader that knows one. Empty
+	// for a walk kept before it was: Repository and Commit are then the whole of it.
+	Commits []PlanCommit `json:"commits,omitempty"`
 }
 
+// PlanCommit is one repository's commit a walk carries: the latest merge of its branch in the batch.
+type PlanCommit struct {
+	Repository string    `json:"repository"`
+	Branch     string    `json:"branch,omitempty"`
+	Commit     string    `json:"commit"`
+	Merged     time.Time `json:"merged,omitzero"`
+}
+
+// PlanMerge is one merge a walk or batch answers (novox/hq ADR 0276), as mesh-delivery reads it: Carried is
+// the commit of its repository the walk builds and sends when that is a later merge containing it, and
+// empty when it is the merge itself.
+type PlanMerge struct {
+	Repository string `json:"repository"`
+	Commit     string `json:"commit"`
+	Carried    string `json:"carried,omitempty"`
+}
+
+// PlanBatch is a batch's window while it is one (novox/hq ADR 0276): when it closes unless another merge
+// comes, when it closes at the latest, and the walk it waits behind once closed.
+type PlanBatch struct {
+	ClosesAt time.Time `json:"closes_at"`
+	AtMost   time.Time `json:"at_most"`
+	Behind   string    `json:"behind,omitempty"`
+}
+
+// CommitOf is the commit of a repository the walk carries; empty when it carries none of it.
+func (p Plan) CommitOf(repository string) string {
+	for _, c := range p.Commits {
+		if strings.EqualFold(c.Repository, repository) {
+			return c.Commit
+		}
+	}
+	if strings.EqualFold(p.Repository, repository) {
+		return p.Commit
+	}
+	return ""
+}
+
+// Carried is every repository's commit the walk carries: Commits, or its one repository and commit.
+func (p Plan) Carried() []PlanCommit {
+	if len(p.Commits) > 0 {
+		return p.Commits
+	}
+	if p.Repository == "" && p.Commit == "" {
+		return nil
+	}
+	return []PlanCommit{{Repository: p.Repository, Branch: p.Branch, Commit: p.Commit, Merged: p.Merged}}
+}
+
+// Named is the walk as a person reads it: each repository at its commit.
+func (p Plan) Named() string {
+	var out []string
+	for _, c := range p.Carried() {
+		out = append(out, c.Repository+" "+short(c.Commit))
+	}
+	return strings.Join(out, ", ")
+}
+
+// Batch says the record is a batch still being assembled or waiting to be cut, not a walk (ADR 0276).
+func (p Plan) Batch() bool { return p.State == PlanAssembling || p.State == PlanQueued }
+
 // PlanDelivery is what a walk waits for and what came of the wait (novox/hq ADR 0239).
 type PlanDelivery struct {
 	// Awaits is who must say the walk may start: the seat whose holder owns the delivery.
@@ -62,6 +129,14 @@ type PlanDelivery struct {
 	// and the delivery reads it as stopped, not as a build that failed.
 	Stopped    string `json:"stopped,omitempty"`
 	StoppedWhy string `json:"stopped_why,omitempty"`
+	// Merges is every merge the walk or batch answers (novox/hq ADR 0276): its own commits, and the earlier
+	// merges of a repository its later commit contains.
+	Merges []PlanMerge `json:"merges,omitempty"`
+	// TakenOverBy names the walk a walk folded before it started was taken over by: its merges are that
+	// walk's now.
+	TakenOverBy string `json:"taken_over_by,omitempty"`
+	// Batch is the window of a batch; nil once it is cut into a walk.
+	Batch *PlanBatch `json:"batch,omitempty"`
 }
 
 // Waiting is whether the walk waits for its delivery's word.
@@ -207,6 +282,10 @@ const (
 	// PlanSuperseded is a plan a newer merge of the same repository and branch took over (novox/hq
 	// issue 254, ADR 0218): what it had not built is in the newer plan, and its note names it.
 	PlanSuperseded = "superseded"
+	// PlanAssembling is a batch whose merge window is open (novox/hq ADR 0276), and PlanQueued one whose
+	// window closed while a walk is open: neither is a walk yet. Cut, a batch keeps its id and is building.
+	PlanAssembling = "assembling"
+	PlanQueued     = "queued"
 )
 
 // Open says whether the plan is still being worked.
@@ -231,7 +310,12 @@ func (i *Inventory) SavePlan(ctx context.Context, p *Plan) error {
 	if err != nil {
 		return err
 	}
-	var release, delivery []byte
+	var release, delivery, commits []byte
+	if len(p.Commits) > 0 {
+		if commits, err = json.Marshal(p.Commits); err != nil {
+			return err
+		}
+	}
 	if p.Release != nil {
 		if release, err = json.Marshal(p.Release); err != nil {
 			return err
@@ -257,16 +341,18 @@ func (i *Inventory) SavePlan(ctx context.Context, p *Plan) error {
 	var revision int64
 	err = tx.QueryRow(ctx,
 		`insert into release_plan (id, repository, commit_hash, created, updated, state, tier, tiers, modules, note,
-		                           branch, tier_entered, revision, epoch, release, delivery, merged_at)
-		 values ($1, $2, $3, $4, now(), $5, $6, $7, $8, $9, $10, $11, 1, $13, $14, $15, $16)
+		                           branch, tier_entered, revision, epoch, release, delivery, merged_at, commits)
+		 values ($1, $2, $3, $4, now(), $5, $6, $7, $8, $9, $10, $11, 1, $13, $14, $15, $16, $17)
 		 on conflict (id) do update set updated = now(), state = excluded.state, tier = excluded.tier,
 		        tiers = excluded.tiers, modules = excluded.modules, note = excluded.note, branch = excluded.branch,
 		        tier_entered = excluded.tier_entered, revision = release_plan.revision + 1, epoch = excluded.epoch,
-		        release = excluded.release, delivery = excluded.delivery
+		        release = excluded.release, delivery = excluded.delivery, repository = excluded.repository,
+		        commit_hash = excluded.commit_hash, merged_at = excluded.merged_at, commits = excluded.commits,
+		        created = excluded.created
 		  where release_plan.revision = $12
 		 returning revision`,
 		p.ID, p.Repository, p.Commit, p.Created, p.State, p.Tier, tiers, modules, p.Note, p.Branch, entered,
-		p.Revision, epoch, release, delivery, mergedAt(p.Merged)).Scan(&revision)
+		p.Revision, epoch, release, delivery, mergedAt(p.Merged), commits).Scan(&revision)
 	if errors.Is(err, pgx.ErrNoRows) {
 		// The row is there and at another revision — moved since this was read, or there already
 		// when this one is new: either way not this writer's to overwrite. (A plan saved before plans
@@ -349,7 +435,7 @@ func (i *Inventory) PlanByID(ctx context.Context, id string) (Plan, error) {
 func (i *Inventory) plans(ctx context.Context, tail string, args ...any) ([]Plan, error) {
 	rows, err := i.store.Pool().Query(ctx,
 		`select id, repository, commit_hash, created, updated, state, tier, tiers, modules, note, branch,
-		        coalesce(tier_entered, created), revision, coalesce(epoch, 0), release, delivery, merged_at
+		        coalesce(tier_entered, created), revision, coalesce(epoch, 0), release, delivery, merged_at, commits
 		   from release_plan `+tail, args...)
 	if err != nil {
 		return nil, err
@@ -358,14 +444,19 @@ func (i *Inventory) plans(ctx context.Context, tail string, args ...any) ([]Plan
 	var out []Plan
 	for rows.Next() {
 		var p Plan
-		var tiers, modules, release, delivery []byte
+		var tiers, modules, release, delivery, commits []byte
 		var epoch int64
 		var merged *time.Time
 		if err := rows.Scan(&p.ID, &p.Repository, &p.Commit, &p.Created, &p.Updated, &p.State,
 			&p.Tier, &tiers, &modules, &p.Note, &p.Branch, &p.TierEntered, &p.Revision, &epoch, &release,
-			&delivery, &merged); err != nil {
+			&delivery, &merged, &commits); err != nil {
 			return nil, err
 		}
+		if len(commits) > 0 {
+			if err := json.Unmarshal(commits, &p.Commits); err != nil {
+				return nil, err
+			}
+		}
 		if merged != nil {
 			p.Merged = merged.UTC()
 		}