743 lines
26 KiB
Go
743 lines
26 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"regexp"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// The gate merges through the forge bridge, never itself (novox/hq ADR 0437 §8, ADR 0441 §5, ADR 0450 §3). It writes into
|
|
// `merges.<group>` the Approve's heads and ONE member at a time, the next only after the bridge's message on FORGE says the
|
|
// one before it merged, and none after a refusal. It re-reads every member's required statuses before it hands a member
|
|
// (ADR 0441 §2). Once every member is merged it moves `source.main` in one compare-and-set write and the mesh's source has
|
|
// the group as one revision. What it announces for each merge (`merged`) names the ask, its asker the teller, and the heads.
|
|
|
|
// MergeStore is the gate's bucket merges.
|
|
type MergeStore interface {
|
|
Get(group string) (*sdk.Merge, error)
|
|
Put(group string, m sdk.Merge) error
|
|
}
|
|
|
|
// SourceStore is the gate's bucket source, one key, moved by compare-and-set.
|
|
type SourceStore interface {
|
|
// Move reads source.main and writes what update makes of it, unless update says nothing is to be written; a write
|
|
// another writer won is read again and tried again.
|
|
Move(update func(prev *sdk.Source) (sdk.Source, bool)) error
|
|
}
|
|
|
|
// kvMerges is merges over the runtime.
|
|
type kvMerges struct{}
|
|
|
|
func (kvMerges) Get(group string) (*sdk.Merge, error) {
|
|
key, err := sdk.MergesKey(group)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
e, err := stdio.State("merges").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
m, err := sdk.DecodeMerge(e.Value)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the merge of %s is not readable: %w", group, err)
|
|
}
|
|
return &m, nil
|
|
}
|
|
|
|
func (kvMerges) Put(group string, m sdk.Merge) error {
|
|
key, err := sdk.MergesKey(group)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
raw, err := sdk.EncodeMerge(m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = stdio.State("merges").Put(key, json.RawMessage(raw))
|
|
return err
|
|
}
|
|
|
|
// kvSource is source over the runtime.
|
|
type kvSource struct{}
|
|
|
|
func (kvSource) Move(update func(prev *sdk.Source) (sdk.Source, bool)) error {
|
|
state := stdio.State("source")
|
|
for range changeAttempts {
|
|
e, err := state.Get(sdk.SourceMainKey)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var prev *sdk.Source
|
|
if e != nil {
|
|
p, err := sdk.DecodeSource(e.Value)
|
|
if err != nil {
|
|
return fmt.Errorf("source.main is not readable: %w", err)
|
|
}
|
|
prev = &p
|
|
}
|
|
next, write := update(prev)
|
|
if !write {
|
|
return nil
|
|
}
|
|
raw, err := sdk.EncodeSource(next)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if e == nil {
|
|
_, err = state.Create(sdk.SourceMainKey, json.RawMessage(raw))
|
|
} else {
|
|
_, err = state.Update(sdk.SourceMainKey, json.RawMessage(raw), e.Revision)
|
|
}
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
if !errors.Is(err, stdio.ErrChanged) {
|
|
return err
|
|
}
|
|
}
|
|
return fmt.Errorf("source.main was written by another writer first %d times: %w", changeAttempts, stdio.ErrChanged)
|
|
}
|
|
|
|
// SourceMain is source.main as the bucket source holds it, read and never written: its group and each repository's trunk
|
|
// head, or why it cannot be read. Outside the lock.
|
|
func (h *Holder) SourceMain() map[string]any {
|
|
if h.Source == nil {
|
|
return map[string]any{"error": "no source bucket"}
|
|
}
|
|
var held *sdk.Source
|
|
err := h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) {
|
|
held = prev
|
|
return sdk.Source{}, false
|
|
})
|
|
switch {
|
|
case err != nil:
|
|
return map[string]any{"error": err.Error()}
|
|
case held == nil:
|
|
return map[string]any{"heads": map[string]string{}, "says": "source.main is not written yet"}
|
|
}
|
|
return map[string]any{"group": held.Group, "heads": held.Heads}
|
|
}
|
|
|
|
// approvedOf are the heads an Approve covers as a merge carries them: `<repository>@<number>` to the full head.
|
|
func (h *Holder) approvedOf(heads map[string]string) map[string]string {
|
|
out := map[string]string{}
|
|
for id, sha := range heads {
|
|
if m := h.deliveries[id]; m != nil {
|
|
out[sdk.ApprovedKey(m.Repository, m.Number)] = sha
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// handTurn is d's turn to be handed to the bridge, when it is: the first member of its group not merged (a lone pull request
|
|
// is a group of one), every member approved at its head and ready to merge. Nil when it is not. Under the lock; the hand
|
|
// runs outside it.
|
|
func (h *Holder) handTurn(g *Group, d *Delivery, now time.Time) func() {
|
|
var reads []memberRead
|
|
c := h.covering(d)
|
|
if g != nil {
|
|
order := h.unmergedLive(g)
|
|
if len(order) == 0 || order[0] != d {
|
|
return nil
|
|
}
|
|
for _, m := range order {
|
|
if h.notHandable(m) != "" {
|
|
return nil // the others are not approved at their heads yet: nothing is merged before they are (said by notHanded)
|
|
}
|
|
reads = append(reads, readOf(m, h.covering(m).Heads[m.ID]))
|
|
}
|
|
}
|
|
if c == nil || c.Approved == nil || c.Heads[d.ID] == "" {
|
|
return nil
|
|
}
|
|
d.Ready.Tries++
|
|
d.Ready.LastTry, d.Ready.merging = now, true
|
|
h.keep(d)
|
|
gid, name, key, target, approved := "", askGroup(d), c.AskID, d.ID, h.approvedOf(c.Heads)
|
|
if g != nil {
|
|
gid = g.ID
|
|
}
|
|
return func() { h.hand(gid, name, key, target, reads, approved) }
|
|
}
|
|
|
|
// notHandable is why a group member cannot be handed in its group's pass, empty when it can: handTurn's own test, in words.
|
|
// Under the lock.
|
|
func (h *Holder) notHandable(m *Delivery) string {
|
|
mc := h.covering(m)
|
|
switch {
|
|
case m.Ready == nil:
|
|
return m.ID + " has no ready run (" + string(m.State) + ")"
|
|
case m.Ready.declined():
|
|
return m.ID + " stands declined since " + m.Ready.Declined.UTC().Format(time.RFC3339)
|
|
case mc == nil || mc.Approved == nil:
|
|
return m.ID + " is not approved at its head " + shortOf(m.Commit, 12)
|
|
case mc.Heads[m.ID] != m.Commit:
|
|
return m.ID + "'s head " + shortOf(m.Commit, 12) + " is not the head approved"
|
|
}
|
|
if w := h.readyWaits(m); w != "" {
|
|
return m.ID + " waits: " + w
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// notHanded is why the approved merge of d, which waits for nothing, is not handed: its group's pass cannot start, and
|
|
// says which member holds it and why. Empty when nothing holds it, or when the pass is under way or its refusal is said
|
|
// already. Under the lock.
|
|
func (h *Holder) notHanded(g *Group, d *Delivery) string {
|
|
if g == nil {
|
|
return ""
|
|
}
|
|
order := h.unmergedLive(g)
|
|
if len(order) == 0 {
|
|
return ""
|
|
}
|
|
if first := order[0].Ready; first != nil && (first.merging || !first.LastTry.IsZero() || first.RefusedSince != nil) {
|
|
return "" // the pass is under way, or its refusal is said on its own line
|
|
}
|
|
for _, m := range order {
|
|
if why := h.notHandable(m); why != "" {
|
|
return fmt.Sprintf("its group %s is not handed to the forge bridge: %s", g.ID, why)
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// noteNotHanded keeps on d's run why its approved merge is not handed, said once in the log when it changes, and as a
|
|
// stalled line meanwhile (readyLines). Under the lock.
|
|
func (h *Holder) noteNotHanded(d *Delivery, why string, now time.Time) {
|
|
r := d.Ready
|
|
if r == nil || r.NotHanded == why {
|
|
return
|
|
}
|
|
r.NotHanded = why
|
|
if why == "" {
|
|
r.NotHandedSince = nil
|
|
} else {
|
|
at := now
|
|
r.NotHandedSince = &at
|
|
h.Logf("[mesh-delivery] %s is approved, waits for nothing, and is NOT handed to merge: %s", d.ID, why)
|
|
}
|
|
h.keep(d)
|
|
}
|
|
|
|
// hand reads a group's required statuses and, when all are a success, writes the target as the member being merged.
|
|
func (h *Holder) hand(gid, name, key, target string, reads []memberRead, approved map[string]string) {
|
|
if gid != "" && !h.groupMergeable(gid, target, reads) {
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
value, ok := h.mergeValue(name, key, approved, target, sdk.MergeMerging)
|
|
h.mu.Unlock()
|
|
if !ok {
|
|
h.giveBack(target, nil)
|
|
return
|
|
}
|
|
cur, err := h.Merges.Get(name)
|
|
if err == nil && cur != nil && sdk.SameMerge(*cur, value) {
|
|
return // handed already, the bridge has the revision
|
|
}
|
|
if err == nil {
|
|
err = h.Merges.Put(name, value)
|
|
}
|
|
if err != nil {
|
|
h.giveBack(target, err)
|
|
return
|
|
}
|
|
h.Logf("[mesh-delivery] %s is handed to the forge bridge to merge, at the heads the operator approved", target)
|
|
}
|
|
|
|
// giveBack gives a target its turn back after a hand that could not be made, to be asked again after a while.
|
|
func (h *Holder) giveBack(target string, err error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if d := h.deliveries[target]; d != nil && d.Ready != nil {
|
|
d.Ready.merging = false
|
|
d.Ready.LastTry = time.Time{}
|
|
}
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s could not be handed to the forge bridge: %v; asked again in %s", target, err, askAgainAfter)
|
|
}
|
|
h.again("a merge that could not be handed to the forge bridge")
|
|
}
|
|
|
|
// mergeValue is the value of a group's merges key with these members merged, and the target in the state asked (merging, or
|
|
// refused with the forge's words): the group's members in order, those the Approve covers. Under the lock. False when it
|
|
// cannot be made (the target is not among the heads the Approve covers).
|
|
func (h *Holder) mergeValue(name, key string, approved map[string]string, target string, state sdk.MergeState) (sdk.Merge, bool) {
|
|
t := h.deliveries[target]
|
|
if t == nil {
|
|
return sdk.Merge{}, false
|
|
}
|
|
members := []*Delivery{t}
|
|
if g := h.groups[t.Group]; t.Group != "" && g != nil {
|
|
members = h.membersOf(g)
|
|
}
|
|
m := sdk.Merge{Group: name, Ask: key, Approved: approved}
|
|
for _, o := range members {
|
|
head := approved[sdk.ApprovedKey(o.Repository, o.Number)]
|
|
if head == "" {
|
|
continue
|
|
}
|
|
pull := sdk.MergePull{Head: sdk.Head{Repository: o.Repository, Number: o.Number, Head: head}}
|
|
switch {
|
|
case o.MergedAs != "" || (o.Ready != nil && o.Ready.Merged != nil):
|
|
commit := o.MergedAs
|
|
if o.Ready != nil && o.Ready.MergeCommit != "" {
|
|
commit = o.Ready.MergeCommit
|
|
}
|
|
if commit == "" {
|
|
continue
|
|
}
|
|
pull.State, pull.Commit = sdk.MergeMerged, commit
|
|
case o == t:
|
|
pull.State = state
|
|
if state == sdk.MergeRefused {
|
|
pull.Words = o.Ready.Answer
|
|
}
|
|
m.Pulls = append(m.Pulls, pull)
|
|
return m, true
|
|
default:
|
|
return m, false
|
|
}
|
|
m.Pulls = append(m.Pulls, pull)
|
|
}
|
|
return m, len(m.Pulls) > 0 && state == sdk.MergeMerged
|
|
}
|
|
|
|
// Forged takes a message of the forge bridge on FORGE: the gate follows a merge and the refusal of a merge, and nothing else.
|
|
func (h *Holder) Forged(event string, body []byte) {
|
|
f, err := sdk.DecodeForge(body)
|
|
if err != nil {
|
|
h.Refuse(fmt.Errorf("a message of the forge bridge is not readable: %v", err))
|
|
return
|
|
}
|
|
switch {
|
|
case f.Kind == sdk.ForgeMerged:
|
|
h.ForgeMerged(f)
|
|
case f.Kind == sdk.ForgeRefused && f.Act == "merge":
|
|
h.ForgeRefused(f)
|
|
}
|
|
}
|
|
|
|
// memberOfForge is the delivery a forge message is about: at the head it names.
|
|
func (h *Holder) memberOfForge(f sdk.Forge) *Delivery {
|
|
if d := h.deliveries[IDOf(f.Repository, f.Head)]; d != nil && d.Number == f.Number {
|
|
return d
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ForgeMerged takes the bridge's word that a pull request the gate handed to it merged: the merge is announced once,
|
|
// with the ask, its asker the teller and the heads (issue 1025), and the next member's turn comes at the next look.
|
|
func (h *Holder) ForgeMerged(f sdk.Forge) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
d := h.memberOfForge(f)
|
|
if d == nil || d.Ready == nil || d.Ready.Approved == nil || d.Ready.Merged != nil {
|
|
return
|
|
}
|
|
r, now := d.Ready, h.Now()
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectEmit, Event: "merged", Line: string(mustJSON(mergedEvent(d, f.Head))), Since: now})
|
|
r.Merged, r.Answer, r.RefusedSince, r.merging, r.MergeCommit = &now, "merged", nil, false, f.Merge
|
|
h.Logf("[mesh-delivery] %s merged on the operator's Approve at %s (%s)", d.ID, shortOf(f.Head, 12), h.checkAt(h.askMembers(d)).Words)
|
|
h.keep(d)
|
|
}
|
|
|
|
// ForgeRefused takes the bridge's word that the forge refused a merge: nothing after it is merged, and the refusal is said
|
|
// as a stalled line (ADR 0441 §3, §4). The merge is handed again only at the next event that decides whether it may
|
|
// merge (mergeAgainNow).
|
|
func (h *Holder) ForgeRefused(f sdk.Forge) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
d := h.memberOfForge(f)
|
|
if d == nil || d.Ready == nil || d.Ready.Merged != nil {
|
|
return
|
|
}
|
|
r, now := d.Ready, h.Now()
|
|
r.merging, r.refusedNoted = false, false
|
|
answer := oneLine(f.Words)
|
|
var on time.Time
|
|
if d.Check != nil {
|
|
on = d.Check.At
|
|
}
|
|
if r.RefusedSince == nil {
|
|
r.RefusedSince, r.FirstRefusedOn = &now, on
|
|
}
|
|
r.LastRefusedOn = on
|
|
r.noteRefusal(d, answer)
|
|
if !outOfDateRefusal(answer) {
|
|
r.Urgent = true // never a warning again for this head
|
|
}
|
|
if r.Repeats >= maxRepeatedRefusals {
|
|
// Loud, not endless (novox/hq issue 1106): the same refusal again with the trunk not moved is no trunk's move, so
|
|
// the merge is not asked again by a verdict or a status; the operator is told, urgent, and decides.
|
|
r.Urgent = true
|
|
if r.Repeats == maxRepeatedRefusals {
|
|
h.Logf("[mesh-delivery] %s: the merge tool refused it %d times in a row with the same words and the trunk not moved, so "+
|
|
"the merge is asked no more until the trunk moves, a new head is announced or the operator approves again: %s",
|
|
d.ID, r.Repeats, answer)
|
|
}
|
|
}
|
|
if answer != r.Answer {
|
|
h.Logf("[mesh-delivery] %s is approved, and the merge tool did not merge it (asked again at the next verdict, Approve or composed check): %s", d.ID,
|
|
answer)
|
|
}
|
|
r.Answer = answer
|
|
h.keep(d)
|
|
}
|
|
|
|
// settleMerges works out, under the lock, what a merge owes its key and the source once the bridge has spoken: a refused
|
|
// member written refused, and a group whose members are all merged written merged and moved onto source.main in one write.
|
|
// The writing happens outside the lock.
|
|
func (h *Holder) settleMerges() []func() {
|
|
var out []func()
|
|
done := map[string]bool{}
|
|
ids := make([]string, 0, len(h.deliveries))
|
|
for id := range h.deliveries {
|
|
ids = append(ids, id)
|
|
}
|
|
sort.Strings(ids)
|
|
for _, id := range ids {
|
|
d := h.deliveries[id]
|
|
r := d.Ready
|
|
if r == nil || r.Approved == nil || r.AskID == "" || len(r.Heads) == 0 {
|
|
continue
|
|
}
|
|
name, key, approved := askGroup(d), r.AskID, h.approvedOf(r.Heads)
|
|
switch {
|
|
case r.Merged == nil && r.RefusedSince != nil && !r.refusedNoted && r.Answer != "" && r.Answer != "merged":
|
|
r.refusedNoted = true
|
|
target := d.ID
|
|
out = append(out, func() {
|
|
h.mu.Lock()
|
|
value, ok := h.mergeValue(name, key, approved, target, sdk.MergeRefused)
|
|
h.mu.Unlock()
|
|
if ok {
|
|
if err := h.Merges.Put(name, value); err != nil {
|
|
h.Logf("[mesh-delivery] the refusal of %s could not be written to its merge: %v", target, err)
|
|
}
|
|
}
|
|
})
|
|
case (r.Merged != nil || d.MergedAs != "") && !r.SourceMoved && !done[key]:
|
|
// Merged through the bridge, or at the forge directly after the Approve (issue 888): the group is moved once.
|
|
if !h.allMerged(r.Heads) {
|
|
continue
|
|
}
|
|
done[key] = true
|
|
heads := map[string]string{}
|
|
var members []string
|
|
for hid := range r.Heads {
|
|
m := h.deliveries[hid]
|
|
if m.LearntLate {
|
|
continue // learnt after the fact: never written to source.main (settleGone)
|
|
}
|
|
commit := m.MergedAs
|
|
if m.Ready != nil && m.Ready.MergeCommit != "" {
|
|
commit = m.Ready.MergeCommit
|
|
}
|
|
heads[m.Repository] = commit
|
|
members = append(members, hid)
|
|
}
|
|
if len(members) == 0 {
|
|
continue // every member learnt after the fact: nothing to write
|
|
}
|
|
sort.Strings(members)
|
|
last := h.deliveries[members[len(members)-1]]
|
|
out = append(out, func() { h.finish(name, key, approved, last.ID, heads, members) })
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// allMerged is whether every delivery an Approve covers is merged.
|
|
func (h *Holder) allMerged(heads map[string]string) bool {
|
|
for id := range heads {
|
|
m := h.deliveries[id]
|
|
if m == nil || (m.MergedAs == "" && (m.Ready == nil || m.Ready.Merged == nil)) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// finish writes a group's merge as all merged and moves source.main to the merge commits in one compare-and-set write: the
|
|
// group is one revision of the mesh's source (ADR 0437 §1, §8).
|
|
func (h *Holder) finish(name, key string, approved map[string]string, last string, heads map[string]string, members []string) {
|
|
h.mu.Lock()
|
|
value, ok := h.mergeValue(name, key, approved, last, sdk.MergeMerged)
|
|
h.mu.Unlock()
|
|
var err error
|
|
if ok {
|
|
err = h.Merges.Put(name, value)
|
|
}
|
|
if err == nil {
|
|
err = h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) {
|
|
next := sdk.Source{Group: name, Heads: map[string]string{}}
|
|
if prev != nil {
|
|
for repository, head := range prev.Heads {
|
|
next.Heads[repository] = head
|
|
}
|
|
}
|
|
for repository, commit := range heads {
|
|
next.Heads[repository] = commit
|
|
}
|
|
return next, prev == nil || !reflect.DeepEqual(prev.Heads, next.Heads)
|
|
})
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] group %s merged, and its place in the source cannot be written yet: %v; asked again in %s", name, err, askAgainAfter)
|
|
h.again("source.main, moved once every member is merged")
|
|
return
|
|
}
|
|
for _, id := range members {
|
|
if m := h.deliveries[id]; m != nil && m.Ready != nil {
|
|
m.Ready.SourceMoved = true
|
|
h.keep(m)
|
|
}
|
|
}
|
|
h.Logf("[mesh-delivery] group %s is merged whole: source.main holds %s", name, strings.Join(members, ", "))
|
|
}
|
|
|
|
// gateMerges says whether finish moves source.main for a delivery: one the operator approved through an ask, whose merge the
|
|
// gate hands to the forge bridge.
|
|
func gateMerges(d *Delivery) bool {
|
|
return d.Ready != nil && d.Ready.Approved != nil && d.Ready.AskID != "" && len(d.Ready.Heads) > 0
|
|
}
|
|
|
|
// leftOutOfForgeSource is whether settleForgeMerges never writes a merge to source.main: one the gate handed on an Approve
|
|
// (finish writes it), and one learnt after the fact, long past or with no time the forge said. One learnt late within
|
|
// replayLate of its merge, as a gate before the fix took mesh-delivery#51 (2026-10-25, its group's finish left it out and
|
|
// source.main kept the trunk before it), is weighed like any merge heard from the forge at a catch-up, and never moves
|
|
// source.main back (olderHead).
|
|
func (d *Delivery) leftOutOfForgeSource() bool {
|
|
if d.LearntLate {
|
|
return d.MergedAt.IsZero() || d.Since.Sub(d.MergedAt) >= replayLate
|
|
}
|
|
return gateMerges(d)
|
|
}
|
|
|
|
// settleForgeMerges moves source.main for each merge into the trunk heard only from the forge (issue 888): a pull request
|
|
// merged at the forge directly, not handed by the gate on an Approve. Every other delivery's trunk moved with it (trunkMoved),
|
|
// and the checker judges on the trunk source.main names, so source.main moves too. A merge replayed as history, or older than
|
|
// a merge of its repository already heard, is never written: source.main never moves backwards. Under the lock; the writing
|
|
// happens outside it.
|
|
func (h *Holder) settleForgeMerges() []func() {
|
|
var out []func()
|
|
again := map[*Delivery]bool{}
|
|
if h.catchUp {
|
|
h.catchUp = false
|
|
again = h.newestInSource()
|
|
}
|
|
for _, d := range h.deliveries {
|
|
if d.MergedAs == "" || (d.InSource && !again[d]) || d.leftOutOfForgeSource() {
|
|
continue
|
|
}
|
|
if d.Base != MergeBranch || d.replayed() || h.newerMergeOf(d) {
|
|
d.InSource = true // not the trunk, or not its head: nothing to write
|
|
h.keep(d)
|
|
continue
|
|
}
|
|
id, repository, base, commit, at, group := d.ID, d.Repository, d.Base, d.MergedAs, mergedWhen(d), firstOf(d.Branch, d.Repository)
|
|
out = append(out, func() {
|
|
err := h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) {
|
|
next := sdk.Source{Group: group, Heads: map[string]string{repository: commit}}
|
|
if prev != nil {
|
|
held := prev.Heads[repository]
|
|
if sameCommit(held, commit) {
|
|
return next, false // there already: the bridge's word and the forge's are one merge
|
|
}
|
|
if held != "" && !h.olderHead(repository, base, held, at) {
|
|
return next, false // source.main holds a newer merge, or one the gate cannot date: never moved back
|
|
}
|
|
for r, head := range prev.Heads {
|
|
if r != repository {
|
|
next.Heads[r] = head
|
|
}
|
|
}
|
|
}
|
|
return next, true
|
|
})
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s merged at the forge, and its place in the source cannot be written yet: %v; asked again in %s", id, err, askAgainAfter)
|
|
h.again("source.main, moved for a merge heard from the forge")
|
|
return
|
|
}
|
|
if m := h.deliveries[id]; m != nil {
|
|
m.InSource = true
|
|
h.keep(m)
|
|
}
|
|
h.Logf("[mesh-delivery] %s merged at the forge: source.main holds %s at %s", id, repository, shortOf(commit, 12))
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// trunkHead is a full commit id as the forge names a branch's head.
|
|
var trunkHead = regexp.MustCompile(`^[0-9a-f]{40}$`)
|
|
|
|
// seedTrunks gives source.main the trunk head of a repository it names none of (issue 1422): one never merged through the
|
|
// gate, whose open pull requests the checker could otherwise never judge. The head is the base_sha of the latest announcement of
|
|
// any open pull request into main (gitea.pull.updated; ADR 0323; lastAnnounced), the trunk's freshest head the gate heard. It is
|
|
// added in one compare-and-set write that never replaces a head source.main holds, so a merge always wins; the first merge
|
|
// through the gate moves it as any other. source.main is per repository and names main's head only (ADR 0484 §11, MergeBranch):
|
|
// a pull request into another base seeds nothing. A repository is weighed once per start after it was found or seeded.
|
|
// Under the lock; the writing happens outside it.
|
|
func (h *Holder) seedTrunks() []func() {
|
|
newest := map[string]*Delivery{}
|
|
for _, d := range h.deliveries {
|
|
if h.seeded[d.Repository] || d.Base != MergeBranch || !trunkHead.MatchString(d.BaseSHA) || d.State.Final() || d.MergedAs != "" {
|
|
continue
|
|
}
|
|
if o := newest[d.Repository]; o == nil || lastAnnounced(d).After(lastAnnounced(o)) {
|
|
newest[d.Repository] = d
|
|
}
|
|
}
|
|
var out []func()
|
|
for repository, d := range newest {
|
|
id, head, group := d.ID, d.BaseSHA, firstOf(d.Branch, d.Repository)
|
|
out = append(out, func() {
|
|
seeded := false
|
|
err := h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) {
|
|
seeded = false
|
|
next := sdk.Source{Group: group, Heads: map[string]string{}}
|
|
if prev != nil {
|
|
if prev.Heads[repository] != "" {
|
|
return *prev, false // named already: never replaced
|
|
}
|
|
next.Group = prev.Group
|
|
for r, held := range prev.Heads {
|
|
next.Heads[r] = held
|
|
}
|
|
}
|
|
next.Heads[repository] = head
|
|
seeded = true
|
|
return next, true
|
|
})
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s: source.main names no trunk head of %s, and it cannot be seeded yet: %v; asked again in %s", id, repository, err, askAgainAfter)
|
|
h.again("source.main, seeded with the trunk head of a repository it names none of")
|
|
return
|
|
}
|
|
if h.seeded == nil {
|
|
h.seeded = map[string]bool{}
|
|
}
|
|
h.seeded[repository] = true
|
|
if seeded {
|
|
h.Logf("[mesh-delivery] source.main named no trunk head of %s: seeded with %s, main's head as the forge announced %s with, so its pull requests are judged (issue 1422)", repository, shortOf(head, 12), id)
|
|
}
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// lastAnnounced is when the forge last announced a delivery's head with its base head: BaseSHAAt, or when the gate made it
|
|
// for one kept before that was written.
|
|
func lastAnnounced(d *Delivery) time.Time {
|
|
if !d.BaseSHAAt.IsZero() {
|
|
return d.BaseSHAAt
|
|
}
|
|
return d.Created
|
|
}
|
|
|
|
// replayLate is how long after its merge a merge heard only from the forge's word is a replay: the forge announces a merge
|
|
// within seconds, so one heard this late was replayed from the bus's history, and the source already holds what came after.
|
|
const replayLate = 15 * time.Minute
|
|
|
|
// replayed is whether the merge was heard from the forge's word replayLate or more after it happened (novox/hq issue 313,
|
|
// kept for source.main's sake after the history state went, ADR 0539 §17): never written to source.main.
|
|
func (d *Delivery) replayed() bool {
|
|
return d.FromForgesWord() && !d.MergedAt.IsZero() && d.Created.Sub(d.MergedAt) >= replayLate
|
|
}
|
|
|
|
// mergeOf is a delivery's merge into its trunk as the gate knows it: the forge's word (MergedAs), or else the bridge's
|
|
// (Ready.MergeCommit, Ready.Merged), which comes first for a merge the gate handed. Empty when it is not merged.
|
|
func mergeOf(d *Delivery) (string, time.Time) {
|
|
if d.MergedAs != "" {
|
|
return d.MergedAs, mergedWhen(d)
|
|
}
|
|
if d.Ready != nil && d.Ready.Merged != nil && d.Ready.MergeCommit != "" {
|
|
return d.Ready.MergeCommit, *d.Ready.Merged
|
|
}
|
|
return "", time.Time{}
|
|
}
|
|
|
|
// newerMergeOf says whether a merge into the same trunk of d's repository was heard that landed after d's.
|
|
func (h *Holder) newerMergeOf(d *Delivery) bool {
|
|
for _, o := range h.deliveries {
|
|
if commit, at := mergeOf(o); o != d && o.Repository == d.Repository && o.Base == d.Base && commit != "" &&
|
|
!sameCommit(commit, d.MergedAs) && at.After(mergedWhen(d)) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// olderHead says whether the head source.main holds for a repository is a merge the gate knows landed before at, so a merge
|
|
// at that time may take its place. A head it does not know (written before its records, or pruned with its delivery) is
|
|
// older only when at is within the time final records are kept: a pruned merge ended before then (issue 888, review).
|
|
// Takes the lock.
|
|
func (h *Holder) olderHead(repository, base, head string, at time.Time) bool {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for _, o := range h.deliveries {
|
|
if commit, when := mergeOf(o); o.Repository == repository && o.Base == base && commit != "" && sameCommit(commit, head) {
|
|
return !when.After(at)
|
|
}
|
|
}
|
|
return at.After(h.Now().Add(-keptFinal))
|
|
}
|
|
|
|
// mergedWhen is when a merged delivery's merge landed: its merge time (the forge's), else when the gate made the delivery. A
|
|
// zero time made every other merge of the repository look newer, so such a merge never moved source.main (issue 1726).
|
|
func mergedWhen(d *Delivery) time.Time {
|
|
if !d.MergedAt.IsZero() {
|
|
return d.MergedAt
|
|
}
|
|
return d.Created
|
|
}
|
|
|
|
// SourceCatchUp asks the next look to weigh again, once, the newest merge into the trunk of each repository though it is
|
|
// marked in the source (issue 1726, mesh-catalog #474, kept here because no seat holder arrives in release 6 to run it): what
|
|
// the gate missed while it was not running, or marked without writing, reaches source.main. The write never moves a repository
|
|
// back (olderHead), and one already there is not written again.
|
|
func (h *Holder) SourceCatchUp() {
|
|
h.mu.Lock()
|
|
h.catchUp = true
|
|
h.mu.Unlock()
|
|
h.Poke()
|
|
}
|
|
|
|
// newestInSource is, for each repository, its newest merge into the trunk heard from the forge, marked in the source. Under the lock.
|
|
func (h *Holder) newestInSource() map[*Delivery]bool {
|
|
newest := map[string]*Delivery{}
|
|
for _, d := range h.deliveries {
|
|
if d.MergedAs == "" || !d.InSource || d.leftOutOfForgeSource() || d.Base != MergeBranch || d.replayed() {
|
|
continue
|
|
}
|
|
if o := newest[d.Repository]; o == nil || mergedWhen(d).After(mergedWhen(o)) {
|
|
newest[d.Repository] = d
|
|
}
|
|
}
|
|
out := map[*Delivery]bool{}
|
|
for _, d := range newest {
|
|
if !h.newerMergeOf(d) {
|
|
out[d] = true
|
|
}
|
|
}
|
|
return out
|
|
}
|