mesh-catalog/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"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)
}

// 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 {
			mc := h.covering(m)
			if m.Ready == nil || m.Ready.Declined != nil || mc == nil || mc.Approved == nil || mc.Heads[m.ID] != m.Commit || h.readyWaits(m) != "" {
				return nil // the others are not approved at their heads yet: nothing is merged before they are
			}
			reads = append(reads, readOf(m, mc.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) }
}

// 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, its check passed against its trunk's head", d.ID, shortOf(f.Head, 12))
	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 && !r.SourceMoved && !done[key]:
			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]
				commit := m.MergedAs
				if m.Ready != nil && m.Ready.MergeCommit != "" {
					commit = m.Ready.MergeCommit
				}
				heads[m.Repository] = commit
				members = append(members, hid)
			}
			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, ", "))
}