internal/control/mergequeue.go
328 lines · 11215 bytes
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}