internal/control/mergequeue.go

  1package control
  2
  3import (
  4	"bytes"
  5	"encoding/json"
  6	"errors"
  7	"fmt"
  8	"io"
  9	"log/slog"
 10	"strconv"
 11	"sync"
 12	"time"
 13
 14	"gitbay.org/gitbay/internal/config"
 15	"gitbay.org/gitbay/internal/policy"
 16	"gitbay.org/gitbay/internal/protocol"
 17	"gitbay.org/gitbay/internal/store"
 18)
 19
 20// QueuedOut is a merge request's queued merge (mr merge --when-ready):
 21// who queued it, the strategy it will use ("" for the default), and why
 22// the last attempt did not merge.
 23type QueuedOut struct {
 24	By       string `json:"by"`
 25	Strategy string `json:"strategy,omitempty"`
 26	Reason   string `json:"reason,omitempty"`
 27	QueuedAt string `json:"queued_at"`
 28}
 29
 30func queuedOut(m store.MR) *QueuedOut {
 31	if m.QueuedAt == "" {
 32		return nil
 33	}
 34	return &QueuedOut{By: m.QueuedBy, Strategy: m.QueueStrategy, Reason: m.QueueReason, QueuedAt: m.QueuedAt}
 35}
 36
 37// mergeQueueMu serialises queued merge attempts, so two triggers landing
 38// together cannot both merge one request.
 39var mergeQueueMu sync.Mutex
 40
 41type queueResult struct {
 42	merged   bool
 43	dequeued bool
 44	reason   string
 45	out      map[string]any // the merge's output, when merged
 46}
 47
 48// TryQueuedMerge attempts the queued merge of one merge request, if it
 49// has one. Called wherever a gate can have changed: a review, a resolved
 50// thread, a draft marked ready.
 51func TryQueuedMerge(st *store.Store, cfg config.Config, mrID int64) {
 52	mergeQueueMu.Lock()
 53	defer mergeQueueMu.Unlock()
 54	attemptLocked(st, cfg, mrID)
 55}
 56
 57// TryQueuedMergesAt attempts the queued merges in a repository whose head
 58// is sha. Called when a status is reported on sha.
 59func TryQueuedMergesAt(st *store.Store, cfg config.Config, repoID int64, sha string) {
 60	ids, err := st.QueuedMRsAtHead(repoID, sha)
 61	if err != nil {
 62		slog.Error("merge queue: listing", "repo", repoID, "err", err)
 63		return
 64	}
 65	for _, id := range ids {
 66		TryQueuedMerge(st, cfg, id)
 67	}
 68}
 69
 70// QueuedMergePushed is post-receive's call for a queued merge request
 71// whose source branch was pushed by pusherID with a key of scope. A push
 72// the target's writers did not make dequeues it, since otherwise the
 73// queuer's authority would merge commits someone else chose: a deploy
 74// key (which can never merge, whoever registered it) or an account that
 75// cannot write to the target. A push that cannot be checked dequeues
 76// too. A push from a writer keeps it queued, and the new head has to
 77// pass on its own.
 78func QueuedMergePushed(st *store.Store, cfg config.Config, mrID, pusherID int64, scope string) {
 79	mergeQueueMu.Lock()
 80	defer mergeQueueMu.Unlock()
 81	mr, err := st.MRByID(mrID)
 82	if err != nil {
 83		slog.Error("merge queue: push", "mr", mrID, "err", err)
 84		st.DequeueMerge(mrID)
 85		return
 86	}
 87	if mr.QueuedAt == "" {
 88		return
 89	}
 90	if policy.IsDeployScope(scope) {
 91		dequeueWithReason(st, mr, "a deploy key pushed, and a deploy key cannot merge")
 92		return
 93	}
 94	const unchecked = "could not check who pushed"
 95	repo, err := st.RepoByID(mr.RepoID)
 96	if err != nil {
 97		slog.Error("merge queue: push", "mr", mrID, "err", err)
 98		dequeueWithReason(st, mr, unchecked)
 99		return
100	}
101	pusher, err := st.UserByID(pusherID)
102	if err != nil {
103		slog.Error("merge queue: push", "mr", mrID, "err", err)
104		dequeueWithReason(st, mr, unchecked)
105		return
106	}
107	grant, err := st.AccessRole(repo.ID, pusher.ID)
108	if err != nil {
109		slog.Error("merge queue: push", "mr", mrID, "err", err)
110		dequeueWithReason(st, mr, unchecked)
111		return
112	}
113	if !policy.CanWrite(pusher, repo, grant) {
114		dequeueWithReason(st, mr, fmt.Sprintf("%s pushed and cannot merge into %s", pusher.Username, repo.Path()))
115		return
116	}
117	attemptLocked(st, cfg, mrID)
118}
119
120// queueSource is Ctx.Source for a merge the queue performs.
121const queueSource = "queue"
122
123// queueInternalReason is what a queued merge says while a lookup fails.
124const queueInternalReason = "internal error, will retry on the next event"
125
126func queueInternalError(st *store.Store, mrID int64, err error) queueResult {
127	slog.Error("merge queue", "mr", mrID, "err", err)
128	st.SetMergeQueueReason(mrID, queueInternalReason)
129	return queueResult{reason: queueInternalReason}
130}
131
132// attemptLocked merges a queued request as the user who queued it,
133// checked against that user's rights and the credential it was queued
134// with now. A merge refused for anything the queuer can fix (unmet
135// gates, a branch behind a require-signed target, a conflict) stays
136// queued with the refusal recorded; a queuer or credential that can no
137// longer merge, or a source branch that is gone, dequeues it. The caller
138// holds mergeQueueMu.
139func attemptLocked(st *store.Store, cfg config.Config, mrID int64) queueResult {
140	mr, err := st.MRByID(mrID)
141	if errors.Is(err, store.ErrNotFound) {
142		return queueResult{}
143	}
144	if err != nil {
145		return queueInternalError(st, mrID, err)
146	}
147	if mr.QueuedAt == "" {
148		return queueResult{}
149	}
150	if mr.State != "open" {
151		return dequeueWithReason(st, mr, "the source branch was deleted")
152	}
153	repo, err := st.RepoByID(mr.RepoID)
154	if err != nil {
155		return queueInternalError(st, mr.ID, err)
156	}
157	user, err := st.UserByID(mr.QueuedByID)
158	if err != nil {
159		return queueInternalError(st, mr.ID, err)
160	}
161	grant, err := st.AccessRole(repo.ID, user.ID)
162	if err != nil {
163		return queueInternalError(st, mr.ID, err)
164	}
165	switch {
166	case user.Disabled || user.Pending:
167		return dequeueWithReason(st, mr, user.Username+"'s account is not active")
168	case !policy.CanWrite(user, repo, grant):
169		return dequeueWithReason(st, mr,
170			fmt.Sprintf("%s no longer has write access to %s", user.Username, repo.Path()))
171	}
172	lapsed, err := queueCredentialLapsed(st, mr.ID, time.Now())
173	if err != nil {
174		return queueInternalError(st, mr.ID, err)
175	}
176	if lapsed != "" {
177		return dequeueWithReason(st, mr, lapsed)
178	}
179
180	var out bytes.Buffer
181	c := &Ctx{User: user, Scope: "full", Source: queueSource, Store: st, Cfg: cfg,
182		Stdin: emptyReader{}, Stdout: &out, Stderr: io.Discard, JSON: true}
183	code := mergeMR(c, repo, mr, mr.QueueStrategy)
184	var env struct {
185		Data  map[string]any `json:"data"`
186		Error string         `json:"error"`
187	}
188	json.Unmarshal(out.Bytes(), &env)
189	if code == protocol.ExitOK {
190		st.Audit(user.ID, "cmd mr merge", map[string]any{
191			"argv": []string{repo.Path(), strconv.FormatInt(mr.Number, 10), "--when-ready"}, "source": queueSource})
192		return queueResult{merged: true, out: env.Data}
193	}
194	st.SetMergeQueueReason(mr.ID, env.Error)
195	return queueResult{reason: env.Error}
196}
197
198// queueCredentialLapsed says why the key or token a merge was queued with
199// can no longer carry it: removed, expired, or narrowed below full scope.
200// "" means it still can, or the merge was queued from a web session and
201// rests on the account alone.
202func queueCredentialLapsed(st *store.Store, mrID int64, now time.Time) (string, error) {
203	q, err := st.MergeQueueCredential(mrID)
204	if err != nil {
205		return "", err
206	}
207	switch q.Kind {
208	case "key":
209		if q.KeyID == 0 {
210			return "the key it was queued with was removed", nil
211		}
212		k, err := st.SSHKeyByID(q.KeyID)
213		if errors.Is(err, store.ErrNotFound) {
214			return "the key it was queued with was removed", nil
215		}
216		if err != nil {
217			return "", err
218		}
219		if k.Expired(now) {
220			return "the key it was queued with has expired", nil
221		}
222		if k.Scope != "full" {
223			return "the key it was queued with no longer has full scope", nil
224		}
225	case "token":
226		if q.TokenID == 0 {
227			return "the token it was queued with was revoked", nil
228		}
229		t, err := st.APITokenByID(q.TokenID)
230		if errors.Is(err, store.ErrNotFound) {
231			return "the token it was queued with was revoked", nil
232		}
233		if err != nil {
234			return "", err
235		}
236		if t.ExpiresAt != nil && !t.ExpiresAt.After(now) {
237			return "the token it was queued with has expired", nil
238		}
239		if t.Scope != "full" {
240			return "the token it was queued with no longer has full scope", nil
241		}
242	}
243	return "", nil
244}
245
246// dequeueWithReason takes mr off the queue and says why on its timeline,
247// as the queuer, whose request it was.
248func dequeueWithReason(st *store.Store, mr store.MR, reason string) queueResult {
249	st.DequeueMerge(mr.ID)
250	st.AddMRSystemComment(mr.ID, mr.QueuedByID, "dequeued the merge queued by "+mr.QueuedBy+": "+reason)
251	return queueResult{dequeued: true, reason: reason}
252}
253
254// queueMerge is mr merge --when-ready: queue the merge as c.User and try
255// it at once, so gates that already pass merge now.
256func queueMerge(c *Ctx, repo store.Repo, mr store.MR, strategy string) int {
257	if code := refuseArchived(c, repo); code >= 0 {
258		return code
259	}
260	if mr.State != "open" {
261		return c.fail(protocol.ExitUsage, "MR !%d is %s", mr.Number, mr.State)
262	}
263	// The merge happens later on this credential's authority, which must
264	// not outlive it (#257).
265	if c.Expires != nil {
266		return c.fail(protocol.ExitDenied,
267			"--when-ready merges later on the authority of the credential it is queued with, and this one expires; queue it with a key or token without an expiry, or from the web")
268	}
269	// A strategy the repository refuses outright would wait forever.
270	if repo.Settings.RequireSignedCommits && (strategy == "merge" || strategy == "squash") {
271		return c.fail(protocol.ExitDenied,
272			"%s requires signed commits, so only fast-forward merges are allowed; queue without --strategy or with --strategy ff", repo.Path())
273	}
274	keyID, code := queueKeyID(c)
275	if code >= 0 {
276		return code
277	}
278	if err := c.Store.QueueMerge(mr.ID, c.User.ID, strategy, keyID, c.TokenID); err != nil {
279		return c.fail(protocol.ExitFailure, "%v", err)
280	}
281	mergeQueueMu.Lock()
282	res := attemptLocked(c.Store, c.Cfg, mr.ID)
283	mergeQueueMu.Unlock()
284	switch {
285	case res.merged:
286		return c.emit(res.out, func(w io.Writer) {
287			fmt.Fprintf(w, "merged %s!%d into %s (%v) at %.10v\n", repo.Path(), mr.Number, mr.TargetRef, res.out["strategy"], res.out["sha"])
288		})
289	case res.dequeued:
290		return c.fail(protocol.ExitDenied, "%s", res.reason)
291	}
292	note := ""
293	if strategy != "" {
294		note = " (" + strategy + ")"
295	}
296	c.Store.AddMRSystemComment(mr.ID, c.User.ID, fmt.Sprintf("%s queued the merge%s for when the gates pass", c.User.Username, note))
297	return c.emit(map[string]any{"number": mr.Number, "queued": true, "strategy": strategy, "reason": res.reason}, func(w io.Writer) {
298		fmt.Fprintf(w, "queued %s!%d to merge when ready; waiting: %s\n", repo.Path(), mr.Number, res.reason)
299	})
300}
301
302// queueKeyID is the SSH key behind c, 0 for a token, a web session, or
303// a context with no credential. -1 as the code means go on.
304func queueKeyID(c *Ctx) (int64, int) {
305	if c.TokenID != 0 || c.Source == "" || c.Source == SourceWeb || c.Source == "api" {
306		return 0, -1
307	}
308	k, err := c.Store.SSHKeyByFingerprint(c.Source)
309	if err != nil {
310		return 0, c.fail(protocol.ExitFailure, "looking up the key behind this session: %v", err)
311	}
312	return k.ID, -1
313}
314
315// cancelQueuedMerge is mr merge --cancel.
316func cancelQueuedMerge(c *Ctx, repo store.Repo, mr store.MR) int {
317	ok, err := c.Store.DequeueMerge(mr.ID)
318	if err != nil {
319		return c.fail(protocol.ExitFailure, "%v", err)
320	}
321	if !ok {
322		return c.fail(protocol.ExitFailure, "!%d is not queued to merge", mr.Number)
323	}
324	c.Store.AddMRSystemComment(mr.ID, c.User.ID, c.User.Username+" cancelled the queued merge")
325	return c.emit(map[string]any{"number": mr.Number, "queued": false}, func(w io.Writer) {
326		fmt.Fprintf(w, "cancelled the queued merge of %s!%d\n", repo.Path(), mr.Number)
327	})
328}