Commit c1f3c195a2
Verified · cmc
Layout: unified · split
internal/control/mergequeue.go +162 −21
| @@ -3,11 +3,13 @@ package control | |||
| 3 | import ( | 3 | import ( |
| 4 | "bytes" | 4 | "bytes" |
| 5 | "encoding/json" | 5 | "encoding/json" |
| 6 | "errors" | ||
| 6 | "fmt" | 7 | "fmt" |
| 7 | "io" | 8 | "io" |
| 8 | "log/slog" | 9 | "log/slog" |
| 9 | "strconv" | 10 | "strconv" |
| 10 | "sync" | 11 | "sync" |
| 12 | "time" | ||
| 11 | 13 | ||
| 12 | "gitbay.org/gitbay/internal/config" | 14 | "gitbay.org/gitbay/internal/config" |
| 13 | "gitbay.org/gitbay/internal/policy" | 15 | "gitbay.org/gitbay/internal/policy" |
| @@ -45,9 +47,11 @@ type queueResult struct { | |||
| 45 | 47 | ||
| 46 | // TryQueuedMerge attempts the queued merge of one merge request, if it | 48 | // TryQueuedMerge attempts the queued merge of one merge request, if it |
| 47 | // has one. Called wherever a gate can have changed: a review, a resolved | 49 | // has one. Called wherever a gate can have changed: a review, a resolved |
| 48 | // thread, a draft marked ready, a push to the source branch. | 50 | // thread, a draft marked ready. |
| 49 | func TryQueuedMerge(st *store.Store, cfg config.Config, mrID int64) { | 51 | func TryQueuedMerge(st *store.Store, cfg config.Config, mrID int64) { |
| 50 | attemptQueuedMerge(st, cfg, mrID) | 52 | mergeQueueMu.Lock() |
| 53 | defer mergeQueueMu.Unlock() | ||
| 54 | attemptLocked(st, cfg, mrID) | ||
| 51 | } | 55 | } |
| 52 | 56 | ||
| 53 | // TryQueuedMergesAt attempts the queued merges in a repository whose head | 57 | // TryQueuedMergesAt attempts the queued merges in a repository whose head |
| @@ -59,44 +63,106 @@ func TryQueuedMergesAt(st *store.Store, cfg config.Config, repoID int64, sha str | |||
| 59 | return | 63 | return |
| 60 | } | 64 | } |
| 61 | for _, id := range ids { | 65 | for _, id := range ids { |
| 62 | attemptQueuedMerge(st, cfg, id) | 66 | TryQueuedMerge(st, cfg, id) |
| 63 | } | 67 | } |
| 64 | } | 68 | } |
| 65 | 69 | ||
| 66 | // attemptQueuedMerge merges a queued request as the user who queued it, | 70 | // QueuedMergePushed is post-receive's call for a queued merge request |
| 67 | // checked against that user's rights now. A merge refused for anything | 71 | // whose source branch pusherID just pushed or deleted. A push from an |
| 68 | // the queuer can fix (unmet gates, a branch behind a require-signed | 72 | // account that cannot write to the target dequeues it: otherwise the |
| 69 | // target, a conflict) stays queued with the refusal recorded; a queuer | 73 | // queuer's authority would merge commits someone else chose. A push from |
| 70 | // who can no longer merge is dequeued. | 74 | // one that can keeps it queued, and the new head has to pass on its own. |
| 71 | func attemptQueuedMerge(st *store.Store, cfg config.Config, mrID int64) queueResult { | 75 | func QueuedMergePushed(st *store.Store, cfg config.Config, mrID, pusherID int64) { |
| 72 | mergeQueueMu.Lock() | 76 | mergeQueueMu.Lock() |
| 73 | defer mergeQueueMu.Unlock() | 77 | defer mergeQueueMu.Unlock() |
| 74 | mr, err := st.MRByID(mrID) | 78 | mr, err := st.MRByID(mrID) |
| 75 | if err != nil || mr.QueuedAt == "" { | 79 | if err != nil || mr.QueuedAt == "" { |
| 76 | return queueResult{} | 80 | return |
| 77 | } | 81 | } |
| 78 | repo, err := st.RepoByID(mr.RepoID) | 82 | repo, err := st.RepoByID(mr.RepoID) |
| 79 | if err != nil { | 83 | if err != nil { |
| 84 | queueInternalError(st, mr.ID, err) | ||
| 85 | return | ||
| 86 | } | ||
| 87 | pusher, err := st.UserByID(pusherID) | ||
| 88 | if err != nil { | ||
| 89 | queueInternalError(st, mr.ID, err) | ||
| 90 | return | ||
| 91 | } | ||
| 92 | grant, err := st.AccessRole(repo.ID, pusher.ID) | ||
| 93 | if err != nil { | ||
| 94 | queueInternalError(st, mr.ID, err) | ||
| 95 | return | ||
| 96 | } | ||
| 97 | if !policy.CanWrite(pusher, repo, grant) { | ||
| 98 | dequeueWithReason(st, mr, fmt.Sprintf("%s pushed and cannot merge into %s", pusher.Username, repo.Path())) | ||
| 99 | return | ||
| 100 | } | ||
| 101 | attemptLocked(st, cfg, mrID) | ||
| 102 | } | ||
| 103 | |||
| 104 | // queueSource is Ctx.Source for a merge the queue performs. | ||
| 105 | const queueSource = "queue" | ||
| 106 | |||
| 107 | // queueInternalReason is what a queued merge says while a lookup fails. | ||
| 108 | const queueInternalReason = "internal error, will retry on the next event" | ||
| 109 | |||
| 110 | func queueInternalError(st *store.Store, mrID int64, err error) queueResult { | ||
| 111 | slog.Error("merge queue", "mr", mrID, "err", err) | ||
| 112 | st.SetMergeQueueReason(mrID, queueInternalReason) | ||
| 113 | return queueResult{reason: queueInternalReason} | ||
| 114 | } | ||
| 115 | |||
| 116 | // attemptLocked merges a queued request as the user who queued it, | ||
| 117 | // checked against that user's rights and the credential it was queued | ||
| 118 | // with now. A merge refused for anything the queuer can fix (unmet | ||
| 119 | // gates, a branch behind a require-signed target, a conflict) stays | ||
| 120 | // queued with the refusal recorded; a queuer or credential that can no | ||
| 121 | // longer merge, or a source branch that is gone, dequeues it. The caller | ||
| 122 | // holds mergeQueueMu. | ||
| 123 | func attemptLocked(st *store.Store, cfg config.Config, mrID int64) queueResult { | ||
| 124 | mr, err := st.MRByID(mrID) | ||
| 125 | if errors.Is(err, store.ErrNotFound) { | ||
| 80 | return queueResult{} | 126 | return queueResult{} |
| 81 | } | 127 | } |
| 82 | user, err := st.UserByID(mr.QueuedByID) | ||
| 83 | if err != nil { | 128 | if err != nil { |
| 129 | return queueInternalError(st, mrID, err) | ||
| 130 | } | ||
| 131 | if mr.QueuedAt == "" { | ||
| 84 | return queueResult{} | 132 | return queueResult{} |
| 85 | } | 133 | } |
| 134 | if mr.State != "open" { | ||
| 135 | return dequeueWithReason(st, mr, "the source branch was deleted") | ||
| 136 | } | ||
| 137 | repo, err := st.RepoByID(mr.RepoID) | ||
| 138 | if err != nil { | ||
| 139 | return queueInternalError(st, mr.ID, err) | ||
| 140 | } | ||
| 141 | user, err := st.UserByID(mr.QueuedByID) | ||
| 142 | if err != nil { | ||
| 143 | return queueInternalError(st, mr.ID, err) | ||
| 144 | } | ||
| 86 | grant, err := st.AccessRole(repo.ID, user.ID) | 145 | grant, err := st.AccessRole(repo.ID, user.ID) |
| 87 | if err != nil { | 146 | if err != nil { |
| 88 | return queueResult{} | 147 | return queueInternalError(st, mr.ID, err) |
| 89 | } | 148 | } |
| 90 | switch { | 149 | switch { |
| 91 | case user.Disabled || user.Pending: | 150 | case user.Disabled || user.Pending: |
| 92 | return dequeueWithReason(st, repo, mr, user, user.Username+"'s account is not active") | 151 | return dequeueWithReason(st, mr, user.Username+"'s account is not active") |
| 93 | case !policy.CanWrite(user, repo, grant): | 152 | case !policy.CanWrite(user, repo, grant): |
| 94 | return dequeueWithReason(st, repo, mr, user, | 153 | return dequeueWithReason(st, mr, |
| 95 | fmt.Sprintf("%s no longer has write access to %s", user.Username, repo.Path())) | 154 | fmt.Sprintf("%s no longer has write access to %s", user.Username, repo.Path())) |
| 96 | } | 155 | } |
| 156 | lapsed, err := queueCredentialLapsed(st, mr.ID, time.Now()) | ||
| 157 | if err != nil { | ||
| 158 | return queueInternalError(st, mr.ID, err) | ||
| 159 | } | ||
| 160 | if lapsed != "" { | ||
| 161 | return dequeueWithReason(st, mr, lapsed) | ||
| 162 | } | ||
| 97 | 163 | ||
| 98 | var out bytes.Buffer | 164 | var out bytes.Buffer |
| 99 | c := &Ctx{User: user, Scope: "full", Source: "queue", Store: st, Cfg: cfg, | 165 | c := &Ctx{User: user, Scope: "full", Source: queueSource, Store: st, Cfg: cfg, |
| 100 | Stdin: emptyReader{}, Stdout: &out, Stderr: io.Discard, JSON: true} | 166 | Stdin: emptyReader{}, Stdout: &out, Stderr: io.Discard, JSON: true} |
| 101 | code := mergeMR(c, repo, mr, mr.QueueStrategy) | 167 | code := mergeMR(c, repo, mr, mr.QueueStrategy) |
| 102 | var env struct { | 168 | var env struct { |
| @@ -106,16 +172,66 @@ func attemptQueuedMerge(st *store.Store, cfg config.Config, mrID int64) queueRes | |||
| 106 | json.Unmarshal(out.Bytes(), &env) | 172 | json.Unmarshal(out.Bytes(), &env) |
| 107 | if code == protocol.ExitOK { | 173 | if code == protocol.ExitOK { |
| 108 | st.Audit(user.ID, "cmd mr merge", map[string]any{ | 174 | st.Audit(user.ID, "cmd mr merge", map[string]any{ |
| 109 | "argv": []string{repo.Path(), strconv.FormatInt(mr.Number, 10), "--when-ready"}, "source": "queue"}) | 175 | "argv": []string{repo.Path(), strconv.FormatInt(mr.Number, 10), "--when-ready"}, "source": queueSource}) |
| 110 | return queueResult{merged: true, out: env.Data} | 176 | return queueResult{merged: true, out: env.Data} |
| 111 | } | 177 | } |
| 112 | st.SetMergeQueueReason(mr.ID, env.Error) | 178 | st.SetMergeQueueReason(mr.ID, env.Error) |
| 113 | return queueResult{reason: env.Error} | 179 | return queueResult{reason: env.Error} |
| 114 | } | 180 | } |
| 115 | 181 | ||
| 116 | func dequeueWithReason(st *store.Store, repo store.Repo, mr store.MR, user store.User, reason string) queueResult { | 182 | // queueCredentialLapsed says why the key or token a merge was queued with |
| 183 | // can no longer carry it: removed, expired, or narrowed below full scope. | ||
| 184 | // "" means it still can, or the merge was queued from a web session and | ||
| 185 | // rests on the account alone. | ||
| 186 | func queueCredentialLapsed(st *store.Store, mrID int64, now time.Time) (string, error) { | ||
| 187 | q, err := st.MergeQueueCredential(mrID) | ||
| 188 | if err != nil { | ||
| 189 | return "", err | ||
| 190 | } | ||
| 191 | switch q.Kind { | ||
| 192 | case "key": | ||
| 193 | if q.KeyID == 0 { | ||
| 194 | return "the key it was queued with was removed", nil | ||
| 195 | } | ||
| 196 | k, err := st.SSHKeyByID(q.KeyID) | ||
| 197 | if errors.Is(err, store.ErrNotFound) { | ||
| 198 | return "the key it was queued with was removed", nil | ||
| 199 | } | ||
| 200 | if err != nil { | ||
| 201 | return "", err | ||
| 202 | } | ||
| 203 | if k.Expired(now) { | ||
| 204 | return "the key it was queued with has expired", nil | ||
| 205 | } | ||
| 206 | if k.Scope != "full" { | ||
| 207 | return "the key it was queued with no longer has full scope", nil | ||
| 208 | } | ||
| 209 | case "token": | ||
| 210 | if q.TokenID == 0 { | ||
| 211 | return "the token it was queued with was revoked", nil | ||
| 212 | } | ||
| 213 | t, err := st.APITokenByID(q.TokenID) | ||
| 214 | if errors.Is(err, store.ErrNotFound) { | ||
| 215 | return "the token it was queued with was revoked", nil | ||
| 216 | } | ||
| 217 | if err != nil { | ||
| 218 | return "", err | ||
| 219 | } | ||
| 220 | if t.ExpiresAt != nil && !t.ExpiresAt.After(now) { | ||
| 221 | return "the token it was queued with has expired", nil | ||
| 222 | } | ||
| 223 | if t.Scope != "full" { | ||
| 224 | return "the token it was queued with no longer has full scope", nil | ||
| 225 | } | ||
| 226 | } | ||
| 227 | return "", nil | ||
| 228 | } | ||
| 229 | |||
| 230 | // dequeueWithReason takes mr off the queue and says why on its timeline, | ||
| 231 | // as the queuer, whose request it was. | ||
| 232 | func dequeueWithReason(st *store.Store, mr store.MR, reason string) queueResult { | ||
| 117 | st.DequeueMerge(mr.ID) | 233 | st.DequeueMerge(mr.ID) |
| 118 | st.AddMRSystemComment(mr.ID, user.ID, "queued merge by "+user.Username+" cancelled: "+reason) | 234 | st.AddMRSystemComment(mr.ID, mr.QueuedByID, "dequeued the merge queued by "+mr.QueuedBy+": "+reason) |
| 119 | return queueResult{dequeued: true, reason: reason} | 235 | return queueResult{dequeued: true, reason: reason} |
| 120 | } | 236 | } |
| 121 | 237 | ||
| @@ -125,18 +241,30 @@ func queueMerge(c *Ctx, repo store.Repo, mr store.MR, strategy string) int { | |||
| 125 | if code := refuseArchived(c, repo); code >= 0 { | 241 | if code := refuseArchived(c, repo); code >= 0 { |
| 126 | return code | 242 | return code |
| 127 | } | 243 | } |
| 128 | if mr.State != "open" && mr.State != "source_gone" { | 244 | if mr.State != "open" { |
| 129 | return c.fail(protocol.ExitUsage, "MR !%d is %s", mr.Number, mr.State) | 245 | return c.fail(protocol.ExitUsage, "MR !%d is %s", mr.Number, mr.State) |
| 130 | } | 246 | } |
| 247 | // The merge happens later on this credential's authority, which must | ||
| 248 | // not outlive it (#257). | ||
| 249 | if c.Expires != nil { | ||
| 250 | return c.fail(protocol.ExitDenied, | ||
| 251 | "--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") | ||
| 252 | } | ||
| 131 | // A strategy the repository refuses outright would wait forever. | 253 | // A strategy the repository refuses outright would wait forever. |
| 132 | if repo.Settings.RequireSignedCommits && (strategy == "merge" || strategy == "squash") { | 254 | if repo.Settings.RequireSignedCommits && (strategy == "merge" || strategy == "squash") { |
| 133 | return c.fail(protocol.ExitDenied, | 255 | return c.fail(protocol.ExitDenied, |
| 134 | "%s requires signed commits, so only fast-forward merges are allowed; queue without --strategy or with --strategy ff", repo.Path()) | 256 | "%s requires signed commits, so only fast-forward merges are allowed; queue without --strategy or with --strategy ff", repo.Path()) |
| 135 | } | 257 | } |
| 136 | if err := c.Store.QueueMerge(mr.ID, c.User.ID, strategy); err != nil { | 258 | keyID, code := queueKeyID(c) |
| 259 | if code >= 0 { | ||
| 260 | return code | ||
| 261 | } | ||
| 262 | if err := c.Store.QueueMerge(mr.ID, c.User.ID, strategy, keyID, c.TokenID); err != nil { | ||
| 137 | return c.fail(protocol.ExitFailure, "%v", err) | 263 | return c.fail(protocol.ExitFailure, "%v", err) |
| 138 | } | 264 | } |
| 139 | res := attemptQueuedMerge(c.Store, c.Cfg, mr.ID) | 265 | mergeQueueMu.Lock() |
| 266 | res := attemptLocked(c.Store, c.Cfg, mr.ID) | ||
| 267 | mergeQueueMu.Unlock() | ||
| 140 | switch { | 268 | switch { |
| 141 | case res.merged: | 269 | case res.merged: |
| 142 | return c.emit(res.out, func(w io.Writer) { | 270 | return c.emit(res.out, func(w io.Writer) { |
| @@ -155,6 +283,19 @@ func queueMerge(c *Ctx, repo store.Repo, mr store.MR, strategy string) int { | |||
| 155 | }) | 283 | }) |
| 156 | } | 284 | } |
| 157 | 285 | ||
| 286 | // queueKeyID is the SSH key behind c, 0 for a token, a web session, or | ||
| 287 | // a context with no credential. -1 as the code means go on. | ||
| 288 | func queueKeyID(c *Ctx) (int64, int) { | ||
| 289 | if c.TokenID != 0 || c.Source == "" || c.Source == SourceWeb || c.Source == "api" { | ||
| 290 | return 0, -1 | ||
| 291 | } | ||
| 292 | k, err := c.Store.SSHKeyByFingerprint(c.Source) | ||
| 293 | if err != nil { | ||
| 294 | return 0, c.fail(protocol.ExitFailure, "looking up the key behind this session: %v", err) | ||
| 295 | } | ||
| 296 | return k.ID, -1 | ||
| 297 | } | ||
| 298 | |||
| 158 | // cancelQueuedMerge is mr merge --cancel. | 299 | // cancelQueuedMerge is mr merge --cancel. |
| 159 | func cancelQueuedMerge(c *Ctx, repo store.Repo, mr store.MR) int { | 300 | func cancelQueuedMerge(c *Ctx, repo store.Repo, mr store.MR) int { |
| 160 | ok, err := c.Store.DequeueMerge(mr.ID) | 301 | ok, err := c.Store.DequeueMerge(mr.ID) |
internal/control/mergequeue_test.go +192 −5
| @@ -8,6 +8,7 @@ import ( | |||
| 8 | "strconv" | 8 | "strconv" |
| 9 | "strings" | 9 | "strings" |
| 10 | "testing" | 10 | "testing" |
| 11 | "time" | ||
| 11 | 12 | ||
| 12 | "gitbay.org/gitbay/internal/config" | 13 | "gitbay.org/gitbay/internal/config" |
| 13 | "gitbay.org/gitbay/internal/protocol" | 14 | "gitbay.org/gitbay/internal/protocol" |
| @@ -74,10 +75,20 @@ func (f *queueFixture) write(name, body string) { | |||
| 74 | 75 | ||
| 75 | // run dispatches argv as u, returning the exit code and stderr. | 76 | // run dispatches argv as u, returning the exit code and stderr. |
| 76 | func (f *queueFixture) run(u store.User, argv ...string) (int, string, string) { | 77 | func (f *queueFixture) run(u store.User, argv ...string) (int, string, string) { |
| 78 | f.t.Helper() | ||
| 79 | return f.runWith(u, nil, argv...) | ||
| 80 | } | ||
| 81 | |||
| 82 | // runWith is run with the Ctx adjusted first, for the credential behind | ||
| 83 | // the request. | ||
| 84 | func (f *queueFixture) runWith(u store.User, adjust func(*Ctx), argv ...string) (int, string, string) { | ||
| 77 | f.t.Helper() | 85 | f.t.Helper() |
| 78 | var out, errOut bytes.Buffer | 86 | var out, errOut bytes.Buffer |
| 79 | c := &Ctx{User: u, Scope: "full", Store: f.st, Stdout: &out, Stderr: &errOut} | 87 | c := &Ctx{User: u, Scope: "full", Store: f.st, Stdout: &out, Stderr: &errOut} |
| 80 | c.Cfg.Server.Root = f.root | 88 | c.Cfg.Server.Root = f.root |
| 89 | if adjust != nil { | ||
| 90 | adjust(c) | ||
| 91 | } | ||
| 81 | code := Dispatch(c, argv) | 92 | code := Dispatch(c, argv) |
| 82 | return code, out.String(), errOut.String() | 93 | return code, out.String(), errOut.String() |
| 83 | } | 94 | } |
| @@ -281,10 +292,10 @@ func TestWhenReadySignedBehindStaysQueued(t *testing.T) { | |||
| 281 | f.git(f.src, "commit", "-q", "-m", "target moves") | 292 | f.git(f.src, "commit", "-q", "-m", "target moves") |
| 282 | f.git(f.src, "push", "-q", f.dir, "main") | 293 | f.git(f.src, "push", "-q", f.dir, "main") |
| 283 | out := f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready") | 294 | out := f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready") |
| 284 | if !strings.Contains(out, "rebase") { | 295 | if !strings.Contains(out, "rebase and push, and the merge stays queued") { |
| 285 | t.Fatalf("output = %q, want the pending reason to name the rebase", out) | 296 | t.Fatalf("output = %q, want the pending reason to name the rebase", out) |
| 286 | } | 297 | } |
| 287 | f.wantQueued("rebase") | 298 | f.wantQueued("is behind main") |
| 288 | 299 | ||
| 289 | // A strategy that cannot ever pass there is refused, not queued. | 300 | // A strategy that cannot ever pass there is refused, not queued. |
| 290 | code, _, errOut := f.run(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready", "--strategy", "squash") | 301 | code, _, errOut := f.run(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready", "--strategy", "squash") |
| @@ -311,9 +322,7 @@ func TestWhenReadyRightsLossDequeues(t *testing.T) { | |||
| 311 | if mr.State != "open" || mr.QueuedAt != "" { | 322 | if mr.State != "open" || mr.QueuedAt != "" { |
| 312 | t.Fatalf("MR = state %s queued_at %q, want open and dequeued", mr.State, mr.QueuedAt) | 323 | t.Fatalf("MR = state %s queued_at %q, want open and dequeued", mr.State, mr.QueuedAt) |
| 313 | } | 324 | } |
| 314 | if sys := f.systemComments(); !strings.Contains(sys, "bob no longer has write access") { | 325 | f.wantDequeued("bob no longer has write access") |
| 315 | t.Fatalf("system comments = %q, want the dequeue reason", sys) | ||
| 316 | } | ||
| 317 | } | 326 | } |
| 318 | 327 | ||
| 319 | // --cancel dequeues; a second cancel has nothing to take off. | 328 | // --cancel dequeues; a second cancel has nothing to take off. |
| @@ -349,3 +358,181 @@ func TestWhenReadyCloseDequeues(t *testing.T) { | |||
| 349 | t.Fatalf("closed MR = %+v, want closed and dequeued", mr) | 358 | t.Fatalf("closed MR = %+v, want closed and dequeued", mr) |
| 350 | } | 359 | } |
| 351 | } | 360 | } |
| 361 | |||
| 362 | // wantDequeued checks !1 is open, off the queue, and its timeline says why. | ||
| 363 | func (f *queueFixture) wantDequeued(why string) { | ||
| 364 | f.t.Helper() | ||
| 365 | mr := f.mr() | ||
| 366 | if mr.State != "open" || mr.QueuedAt != "" { | ||
| 367 | f.t.Fatalf("MR = state %s queued_at %q, want open and dequeued", mr.State, mr.QueuedAt) | ||
| 368 | } | ||
| 369 | if sys := f.systemComments(); !strings.Contains(sys, "dequeued the merge queued by") || !strings.Contains(sys, why) { | ||
| 370 | f.t.Fatalf("system comments = %q, want a dequeue saying %q", sys, why) | ||
| 371 | } | ||
| 372 | } | ||
| 373 | |||
| 374 | // branch points a new branch of the bare repository at sha. | ||
| 375 | func (f *queueFixture) branch(name, sha string) { | ||
| 376 | f.t.Helper() | ||
| 377 | f.git(f.dir, "update-ref", "refs/heads/"+name, sha) | ||
| 378 | } | ||
| 379 | |||
| 380 | // Retargeting a queued merge request by someone who cannot write to the | ||
| 381 | // repository dequeues it; by someone who can, it stays queued. | ||
| 382 | func TestWhenReadyRetargetByNonWriterDequeues(t *testing.T) { | ||
| 383 | f := newQueueFixture(t, func(s *store.RepoSettings) { s.RequireApprovals = 1 }) | ||
| 384 | f.branch("dev", f.targetSH) | ||
| 385 | f.branch("dev2", f.targetSH) | ||
| 386 | // carol authored it and can read, so she may retarget it. | ||
| 387 | carol := f.user("carol", "read") | ||
| 388 | if _, err := f.st.DB.Exec("UPDATE merge_requests SET author_id = ? WHERE id = ?", carol.ID, f.mr().ID); err != nil { | ||
| 389 | t.Fatal(err) | ||
| 390 | } | ||
| 391 | f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready") | ||
| 392 | f.mustRun(f.alice, "mr", "retarget", f.repo.Path(), "1", "dev") | ||
| 393 | f.wantQueued("approval") | ||
| 394 | f.mustRun(carol, "mr", "retarget", f.repo.Path(), "1", "dev2") | ||
| 395 | f.wantDequeued("carol retargeted it to dev2 and cannot merge") | ||
| 396 | } | ||
| 397 | |||
| 398 | // The stack moving up after a merge retargets the merge requests on it, | ||
| 399 | // and a queued one stays queued. | ||
| 400 | func TestWhenReadyStackRetargetKeepsQueue(t *testing.T) { | ||
| 401 | f := newQueueFixture(t, func(s *store.RepoSettings) { | ||
| 402 | s.RequireChecks = true | ||
| 403 | s.RequiredContexts = []string{"ext/test"} | ||
| 404 | }) | ||
| 405 | f.write("stacked.txt", "s\n") | ||
| 406 | f.git(f.src, "add", ".") | ||
| 407 | f.git(f.src, "commit", "-q", "-m", "stacked") | ||
| 408 | stacked := strings.TrimSpace(f.git(f.src, "rev-parse", "HEAD")) | ||
| 409 | f.git(f.src, "push", "-q", f.dir, "HEAD:refs/heads/feature2") | ||
| 410 | f.git(f.dir, "update-ref", mrHeadRef(2), stacked) | ||
| 411 | if _, err := f.st.CreateMR(f.repo.ID, f.alice.ID, f.repo.ID, "feature2", "feature", "two", "", stacked, "md", false); err != nil { | ||
| 412 | t.Fatal(err) | ||
| 413 | } | ||
| 414 | f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "2", "--when-ready") | ||
| 415 | f.mustRun(f.alice, "status", "set", f.repo.Path(), f.headSHA, "--context", "ext/test", "--state", "success") | ||
| 416 | f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "1") | ||
| 417 | two, err := f.st.MRByNumber(f.repo.ID, 2) | ||
| 418 | if err != nil { | ||
| 419 | t.Fatal(err) | ||
| 420 | } | ||
| 421 | if two.TargetRef != "main" || two.State != "open" || two.QueuedAt == "" { | ||
| 422 | t.Fatalf("!2 after !1 merged = target %s state %s queued_at %q, want main, open and queued", two.TargetRef, two.State, two.QueuedAt) | ||
| 423 | } | ||
| 424 | } | ||
| 425 | |||
| 426 | // A merge queued with an SSH key is dequeued when the key is removed. | ||
| 427 | func TestWhenReadyRemovedKeyDequeues(t *testing.T) { | ||
| 428 | f := newQueueFixture(t, func(s *store.RepoSettings) { s.RequireApprovals = 1 }) | ||
| 429 | if err := f.st.AddSSHKey(f.alice.ID, "SHA256:alice", "ssh-ed25519", []byte("a"), "full", ""); err != nil { | ||
| 430 | t.Fatal(err) | ||
| 431 | } | ||
| 432 | withKey := func(c *Ctx) { c.Source = "SHA256:alice" } | ||
| 433 | if code, _, errOut := f.runWith(f.alice, withKey, "mr", "merge", f.repo.Path(), "1", "--when-ready"); code != protocol.ExitOK { | ||
| 434 | t.Fatalf("queue: exit %d, %s", code, errOut) | ||
| 435 | } | ||
| 436 | if err := f.st.RemoveSSHKey(f.alice.ID, "SHA256:alice"); err != nil { | ||
| 437 | t.Fatal(err) | ||
| 438 | } | ||
| 439 | f.mustRun(f.user("bob", "write"), "mr", "review", f.repo.Path(), "1", "--approve") | ||
| 440 | f.wantDequeued("the key it was queued with was removed") | ||
| 441 | } | ||
| 442 | |||
| 443 | // A merge queued with an API token is dequeued when the token is revoked. | ||
| 444 | func TestWhenReadyRevokedTokenDequeues(t *testing.T) { | ||
| 445 | f := newQueueFixture(t, func(s *store.RepoSettings) { s.RequireApprovals = 1 }) | ||
| 446 | if err := f.st.CreateAPIToken(f.alice.ID, "ci", "hash", "full", nil, 0); err != nil { | ||
| 447 | t.Fatal(err) | ||
| 448 | } | ||
| 449 | toks, err := f.st.ListAPITokens(f.alice.ID) | ||
| 450 | if err != nil || len(toks) != 1 { | ||
| 451 | t.Fatalf("tokens = %v, %v", toks, err) | ||
| 452 | } | ||
| 453 | withToken := func(c *Ctx) { c.Source, c.TokenID, c.ViaAPI = "api", toks[0].ID, true } | ||
| 454 | if code, _, errOut := f.runWith(f.alice, withToken, "mr", "merge", f.repo.Path(), "1", "--when-ready"); code != protocol.ExitOK { | ||
| 455 | t.Fatalf("queue: exit %d, %s", code, errOut) | ||
| 456 | } | ||
| 457 | if _, err := f.st.RevokeAPIToken(f.alice.ID, "ci", false); err != nil { | ||
| 458 | t.Fatal(err) | ||
| 459 | } | ||
| 460 | f.mustRun(f.user("bob", "write"), "mr", "review", f.repo.Path(), "1", "--approve") | ||
| 461 | f.wantDequeued("the token it was queued with was revoked") | ||
| 462 | } | ||
| 463 | |||
| 464 | // An expiring credential cannot queue a merge: the merge would happen on | ||
| 465 | // its authority after it lapsed. | ||
| 466 | func TestWhenReadyExpiringCredentialRefused(t *testing.T) { | ||
| 467 | f := newQueueFixture(t, func(s *store.RepoSettings) { s.RequireApprovals = 1 }) | ||
| 468 | soon := time.Now().Add(time.Hour) | ||
| 469 | code, _, errOut := f.runWith(f.alice, func(c *Ctx) { c.Source, c.TokenID, c.Expires = "api", 7, &soon }, | ||
| 470 | "mr", "merge", f.repo.Path(), "1", "--when-ready") | ||
| 471 | if code != protocol.ExitDenied || !strings.Contains(errOut, "without an expiry, or from the web") { | ||
| 472 | t.Fatalf("exit %d, %s", code, errOut) | ||
| 473 | } | ||
| 474 | if mr := f.mr(); mr.QueuedAt != "" { | ||
| 475 | t.Fatalf("expiring credential queued the merge: %+v", mr) | ||
| 476 | } | ||
| 477 | } | ||
| 478 | |||
| 479 | // A merge queued from the web rests on the account alone and merges. | ||
| 480 | func TestWhenReadyWebQueuedMerges(t *testing.T) { | ||
| 481 | f := newQueueFixture(t, func(s *store.RepoSettings) { s.RequireApprovals = 1 }) | ||
| 482 | web := func(c *Ctx) { c.Source, c.ViaAPI = SourceWeb, true } | ||
| 483 | if code, _, errOut := f.runWith(f.alice, web, "mr", "merge", f.repo.Path(), "1", "--when-ready"); code != protocol.ExitOK { | ||
| 484 | t.Fatalf("queue: exit %d, %s", code, errOut) | ||
| 485 | } | ||
| 486 | f.mustRun(f.user("bob", "write"), "mr", "review", f.repo.Path(), "1", "--approve") | ||
| 487 | f.wantMergedBy("alice") | ||
| 488 | } | ||
| 489 | |||
| 490 | // runner done reporting the last required check green merges the queued | ||
| 491 | // merge request. | ||
| 492 | func TestWhenReadyRunnerDoneMerges(t *testing.T) { | ||
| 493 | f := newQueueFixture(t, func(s *store.RepoSettings) { | ||
| 494 | s.RequireChecks = true | ||
| 495 | s.RequiredContexts = []string{"ci/unit"} | ||
| 496 | }) | ||
| 497 | f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready") | ||
| 498 | f.wantQueued("ci/unit=missing") | ||
| 499 | if _, err := f.st.CreateBuild(f.repo.ID, "unit", f.headSHA, "feature", `["true"]`, "", "", true); err != nil { | ||
| 500 | t.Fatal(err) | ||
| 501 | } | ||
| 502 | b, ok, err := f.st.ClaimBuild([]int64{f.repo.ID}, false) | ||
| 503 | if err != nil || !ok { | ||
| 504 | t.Fatalf("claim: ok=%v err=%v", ok, err) | ||
| 505 | } | ||
| 506 | c, out := runnerCtx(f.st, f.alice.ID, f.root) | ||
| 507 | if code := runRunnerDone(c, []string{strconv.FormatInt(b.ID, 10), "success"}); code != protocol.ExitOK { | ||
| 508 | t.Fatalf("runner done: exit %d\n%s", code, out.String()) | ||
| 509 | } | ||
| 510 | f.wantMergedBy("alice") | ||
| 511 | } | ||
| 512 | |||
| 513 | // Cancelling a build whose job already passed on the same commit puts | ||
| 514 | // the success back, and that merges the queued merge request. | ||
| 515 | func TestWhenReadyCancelledBuildSuccessMerges(t *testing.T) { | ||
| 516 | f := newQueueFixture(t, func(s *store.RepoSettings) { | ||
| 517 | s.RequireChecks = true | ||
| 518 | s.RequiredContexts = []string{"ci/unit"} | ||
| 519 | }) | ||
| 520 | if _, err := f.st.CreateBuild(f.repo.ID, "unit", f.headSHA, "feature", `["true"]`, "", "", true); err != nil { | ||
| 521 | t.Fatal(err) | ||
| 522 | } | ||
| 523 | b, ok, err := f.st.ClaimBuild([]int64{f.repo.ID}, false) | ||
| 524 | if err != nil || !ok { | ||
| 525 | t.Fatalf("claim: ok=%v err=%v", ok, err) | ||
| 526 | } | ||
| 527 | if err := f.st.FinishBuild(b.ID, "success"); err != nil { | ||
| 528 | t.Fatal(err) | ||
| 529 | } | ||
| 530 | n, err := f.st.CreateBuild(f.repo.ID, "unit", f.headSHA, "other", `["true"]`, "", "", true) | ||
| 531 | if err != nil { | ||
| 532 | t.Fatal(err) | ||
| 533 | } | ||
| 534 | f.mustRun(f.alice, "mr", "merge", f.repo.Path(), "1", "--when-ready") | ||
| 535 | f.wantQueued("ci/unit=missing") | ||
| 536 | f.mustRun(f.alice, "build", "cancel", f.repo.Path(), strconv.FormatInt(n, 10)) | ||
| 537 | f.wantMergedBy("alice") | ||
| 538 | } | ||
internal/control/mr.go +15
| @@ -1047,6 +1047,16 @@ func runMRRetarget(c *Ctx, args []string) int { | |||
| 1047 | return c.fail(protocol.ExitFailure, "%v", err) | 1047 | return c.fail(protocol.ExitFailure, "%v", err) |
| 1048 | } | 1048 | } |
| 1049 | c.Store.AddMRSystemComment(mr.ID, c.User.ID, fmt.Sprintf("retargeted from %s to %s", old, target)) | 1049 | c.Store.AddMRSystemComment(mr.ID, c.User.ID, fmt.Sprintf("retargeted from %s to %s", old, target)) |
| 1050 | // A queued merge was asked for against the old target. Someone who | ||
| 1051 | // could not merge it themselves does not get to point it elsewhere. | ||
| 1052 | if mr.QueuedAt != "" { | ||
| 1053 | grant, err := c.Store.AccessRole(repo.ID, c.User.ID) | ||
| 1054 | if err != nil || !policy.CanWrite(c.User, repo, grant) { | ||
| 1055 | mergeQueueMu.Lock() | ||
| 1056 | dequeueWithReason(c.Store, mr, fmt.Sprintf("%s retargeted it to %s and cannot merge into %s", c.User.Username, target, repo.Path())) | ||
| 1057 | mergeQueueMu.Unlock() | ||
| 1058 | } | ||
| 1059 | } | ||
| 1050 | c.Store.RecordEvent(repo.ID, c.User.ID, "mr.retargeted", | 1060 | c.Store.RecordEvent(repo.ID, c.User.ID, "mr.retargeted", |
| 1051 | fmt.Sprintf(`{"number":%d,"from":%q,"to":%q}`, mr.Number, old, target)) | 1061 | fmt.Sprintf(`{"number":%d,"from":%q,"to":%q}`, mr.Number, old, target)) |
| 1052 | if parts, err := c.Store.MRParticipants(mr.ID); err == nil { | 1062 | if parts, err := c.Store.MRParticipants(mr.ID); err == nil { |
| @@ -1357,6 +1367,11 @@ func mergeMR(c *Ctx, repo store.Repo, mr store.MR, strategy string) int { | |||
| 1357 | // rebase when fast-forward is already possible IS a fast-forward | 1367 | // rebase when fast-forward is already possible IS a fast-forward |
| 1358 | // (nothing is rewritten), so it stays legal. | 1368 | // (nothing is rewritten), so it stays legal. |
| 1359 | if repo.Settings.RequireSignedCommits { | 1369 | if repo.Settings.RequireSignedCommits { |
| 1370 | if c.Source == queueSource && !ffPossible { | ||
| 1371 | return c.fail(protocol.ExitDenied, | ||
| 1372 | "%s is behind %s and %s requires signed commits, so the server cannot rebase it; rebase and push, and the merge stays queued", | ||
| 1373 | mr.SourceRef, mr.TargetRef, repo.Path()) | ||
| 1374 | } | ||
| 1360 | if strategy == "merge" || strategy == "squash" || !ffPossible { | 1375 | if strategy == "merge" || strategy == "squash" || !ffPossible { |
| 1361 | return c.fail(protocol.ExitDenied, | 1376 | return c.fail(protocol.ExitDenied, |
| 1362 | "%s requires signed commits, so only fast-forward merges are allowed; rebase %s onto %s locally, re-push, and merge again", | 1377 | "%s requires signed commits, so only fast-forward merges are allowed; rebase %s onto %s locally, re-push, and merge again", |
internal/hookd/hookd.go +6 −3
| @@ -336,6 +336,9 @@ func (s *Server) postReceive(req Request) { | |||
| 336 | if mr.State == "open" { | 336 | if mr.State == "open" { |
| 337 | s.st.SetMRState(mr.ID, "source_gone") | 337 | s.st.SetMRState(mr.ID, "source_gone") |
| 338 | } | 338 | } |
| 339 | if mr.QueuedAt != "" { | ||
| 340 | control.TryQueuedMerge(s.st, s.cfg, mr.ID) // dequeues: the source is gone | ||
| 341 | } | ||
| 339 | continue // head ref retained: the diff stays viewable | 342 | continue // head ref retained: the diff stays viewable |
| 340 | } | 343 | } |
| 341 | dstDir := control.RepoDir(s.cfg.Server.Root, target.OwnerName, target.Name) | 344 | dstDir := control.RepoDir(s.cfg.Server.Root, target.OwnerName, target.Name) |
| @@ -362,10 +365,10 @@ func (s *Server) postReceive(req Request) { | |||
| 362 | if mr.State == "source_gone" { | 365 | if mr.State == "source_gone" { |
| 363 | s.st.SetMRState(mr.ID, "open") // branch came back | 366 | s.st.SetMRState(mr.ID, "open") // branch came back |
| 364 | } | 367 | } |
| 365 | // A queued merge stays queued across a push; the new head has | 368 | // A queued merge stays queued across a push by someone who can |
| 366 | // to pass the gates on its own. | 369 | // merge it, and the new head has to pass the gates on its own. |
| 367 | if mr.QueuedAt != "" { | 370 | if mr.QueuedAt != "" { |
| 368 | control.TryQueuedMerge(s.st, s.cfg, mr.ID) | 371 | control.QueuedMergePushed(s.st, s.cfg, mr.ID, req.UserID) |
| 369 | } | 372 | } |
| 370 | } | 373 | } |
| 371 | } | 374 | } |
internal/hookd/mergequeue_test.go +110
| @@ -94,3 +94,113 @@ func TestPostReceiveTriesQueuedMerge(t *testing.T) { | |||
| 94 | t.Fatalf("main = %s, want %s", main, rebased) | 94 | t.Fatalf("main = %s, want %s", main, rebased) |
| 95 | } | 95 | } |
| 96 | } | 96 | } |
| 97 | |||
| 98 | // A fork author who cannot write to the target pushes to the source of a | ||
| 99 | // merge request someone else queued: the queue is not theirs to use, so | ||
| 100 | // it is dequeued rather than tried. Deleting the source branch dequeues | ||
| 101 | // it too. | ||
| 102 | func TestPostReceiveDequeuesQueuedMerge(t *testing.T) { | ||
| 103 | st, err := store.Open(":memory:") | ||
| 104 | if err != nil { | ||
| 105 | t.Fatal(err) | ||
| 106 | } | ||
| 107 | t.Cleanup(func() { st.Close() }) | ||
| 108 | if err := st.MigrateUp(); err != nil { | ||
| 109 | t.Fatal(err) | ||
| 110 | } | ||
| 111 | alice, _ := st.CreateUser("alice", false) | ||
| 112 | bob, _ := st.CreateUser("bob", false) | ||
| 113 | targetID, _ := st.CreateRepo("user", alice, "app", "public") | ||
| 114 | forkID, _ := st.CreateRepo("user", bob, "app", "public") | ||
| 115 | target, _ := st.RepoByID(targetID) | ||
| 116 | fork, _ := st.RepoByID(forkID) | ||
| 117 | if _, err := st.UpdateRepoSettings(target.ID, func(s *store.RepoSettings) { s.RequireApprovals = 1 }); err != nil { | ||
| 118 | t.Fatal(err) | ||
| 119 | } | ||
| 120 | root := t.TempDir() | ||
| 121 | f := &shapeFixture{t: t, st: st, repo: fork, uid: bob, root: root, src: filepath.Join(root, "src")} | ||
| 122 | cfg := config.Config{} | ||
| 123 | cfg.Server.Root, cfg.Server.SiteURL = root, "https://x.test" | ||
| 124 | srv := &Server{cfg: cfg, st: st} | ||
| 125 | |||
| 126 | os.MkdirAll(f.src, 0o755) | ||
| 127 | f.git(root, "init", "-q", "-b", "main", "src") | ||
| 128 | f.write("README", "x\n") | ||
| 129 | f.git(f.src, "add", ".") | ||
| 130 | f.git(f.src, "commit", "-q", "-m", "base") | ||
| 131 | f.git(f.src, "checkout", "-q", "-b", "feature") | ||
| 132 | f.write("feature.txt", "y\n") | ||
| 133 | f.git(f.src, "add", ".") | ||
| 134 | f.git(f.src, "commit", "-q", "-m", "change") | ||
| 135 | head := f.sha("HEAD") | ||
| 136 | for _, r := range []store.Repo{target, fork} { | ||
| 137 | dir := control.RepoDir(root, r.OwnerName, r.Name) | ||
| 138 | os.MkdirAll(filepath.Dir(dir), 0o755) | ||
| 139 | f.git(root, "init", "-q", "--bare", dir) | ||
| 140 | f.git(f.src, "push", "-q", dir, "main", "feature") | ||
| 141 | } | ||
| 142 | targetDir := control.RepoDir(root, target.OwnerName, target.Name) | ||
| 143 | f.git(targetDir, "update-ref", "refs/merge-requests/1/head", head) | ||
| 144 | f.git(targetDir, "update-ref", "refs/merge-requests/2/head", head) | ||
| 145 | for range 2 { | ||
| 146 | if _, err := st.CreateMR(target.ID, bob, fork.ID, "feature", "main", "t", "", head, "md", false); err != nil { | ||
| 147 | t.Fatal(err) | ||
| 148 | } | ||
| 149 | } | ||
| 150 | for _, n := range []string{"1", "2"} { | ||
| 151 | var out, errOut bytes.Buffer | ||
| 152 | c := &control.Ctx{User: store.User{ID: alice, Username: "alice"}, Scope: "full", Store: st, Cfg: cfg, Stdout: &out, Stderr: &errOut} | ||
| 153 | if code := control.Dispatch(c, []string{"mr", "merge", target.Path(), n, "--when-ready"}); code != protocol.ExitOK { | ||
| 154 | t.Fatalf("queue !%s: exit %d, %s", n, code, errOut.String()) | ||
| 155 | } | ||
| 156 | } | ||
| 157 | systemSays := func(n int64, want string) { | ||
| 158 | t.Helper() | ||
| 159 | mr, err := st.MRByNumber(target.ID, n) | ||
| 160 | if err != nil { | ||
| 161 | t.Fatal(err) | ||
| 162 | } | ||
| 163 | if mr.QueuedAt != "" || mr.State == "merged" { | ||
| 164 | t.Fatalf("!%d = state %s queued_at %q, want dequeued and not merged", n, mr.State, mr.QueuedAt) | ||
| 165 | } | ||
| 166 | cs, _ := st.ListMRComments(mr.ID) | ||
| 167 | for _, c := range cs { | ||
| 168 | if c.Kind == "system" && strings.Contains(c.Body, want) { | ||
| 169 | return | ||
| 170 | } | ||
| 171 | } | ||
| 172 | t.Fatalf("!%d timeline does not say %q: %+v", n, want, cs) | ||
| 173 | } | ||
| 174 | |||
| 175 | // !2 is closed first so the push reaches only !1. | ||
| 176 | st.MarkClosed(mustMR(t, st, target.ID, 2).ID, alice, "") | ||
| 177 | f.write("feature.txt", "z\n") | ||
| 178 | f.git(f.src, "commit", "-q", "-am", "more") | ||
| 179 | pushed := f.sha("HEAD") | ||
| 180 | forkDir := control.RepoDir(root, fork.OwnerName, fork.Name) | ||
| 181 | f.git(f.src, "push", "-q", forkDir, "feature") | ||
| 182 | srv.postReceive(Request{RepoID: fork.ID, UserID: bob, Scope: "full", Updates: []policy.RefUpdate{ | ||
| 183 | {Ref: "refs/heads/feature", Old: head, New: pushed}}}) | ||
| 184 | systemSays(1, "bob pushed and cannot merge into alice/app") | ||
| 185 | |||
| 186 | // Deleting the source dequeues a queued request. | ||
| 187 | st.SetMRState(mustMR(t, st, target.ID, 2).ID, "open") | ||
| 188 | var out, errOut bytes.Buffer | ||
| 189 | c := &control.Ctx{User: store.User{ID: alice, Username: "alice"}, Scope: "full", Store: st, Cfg: cfg, Stdout: &out, Stderr: &errOut} | ||
| 190 | if code := control.Dispatch(c, []string{"mr", "merge", target.Path(), "2", "--when-ready"}); code != protocol.ExitOK { | ||
| 191 | t.Fatalf("queue !2: exit %d, %s", code, errOut.String()) | ||
| 192 | } | ||
| 193 | f.git(forkDir, "update-ref", "-d", "refs/heads/feature") | ||
| 194 | srv.postReceive(Request{RepoID: fork.ID, UserID: bob, Scope: "full", Updates: []policy.RefUpdate{ | ||
| 195 | {Ref: "refs/heads/feature", Old: pushed, New: zeroSHA40, IsDelete: true}}}) | ||
| 196 | systemSays(2, "the source branch was deleted") | ||
| 197 | } | ||
| 198 | |||
| 199 | func mustMR(t *testing.T, st *store.Store, repoID, n int64) store.MR { | ||
| 200 | t.Helper() | ||
| 201 | mr, err := st.MRByNumber(repoID, n) | ||
| 202 | if err != nil { | ||
| 203 | t.Fatal(err) | ||
| 204 | } | ||
| 205 | return mr | ||
| 206 | } | ||
internal/store/mergequeue.go +37 −7
| @@ -16,18 +16,48 @@ func (s *Store) MRByID(id int64) (MR, error) { | |||
| 16 | } | 16 | } |
| 17 | 17 | ||
| 18 | // QueueMerge queues a merge request to merge as userID with strategy | 18 | // QueueMerge queues a merge request to merge as userID with strategy |
| 19 | // ("" for the default) once its gates pass. Queueing again replaces the | 19 | // ("" for the default) once its gates pass, bound to the SSH key or API |
| 20 | // queuer and strategy and clears the recorded reason. | 20 | // token it was queued with (both 0 for a web session). Queueing again |
| 21 | func (s *Store) QueueMerge(mrID, userID int64, strategy string) error { | 21 | // replaces the queuer, strategy and credential and clears the reason. |
| 22 | func (s *Store) QueueMerge(mrID, userID int64, strategy string, keyID, tokenID int64) error { | ||
| 23 | credential := "" | ||
| 24 | switch { | ||
| 25 | case keyID != 0: | ||
| 26 | credential = "key" | ||
| 27 | case tokenID != 0: | ||
| 28 | credential = "token" | ||
| 29 | } | ||
| 22 | _, err := s.DB.Exec(` | 30 | _, err := s.DB.Exec(` |
| 23 | INSERT INTO mr_merge_queue (mr_id, user_id, strategy) VALUES (?, ?, ?) | 31 | INSERT INTO mr_merge_queue (mr_id, user_id, strategy, credential, key_id, token_id) |
| 32 | VALUES (?, ?, ?, ?, ?, ?) | ||
| 24 | ON CONFLICT (mr_id) DO UPDATE SET user_id = excluded.user_id, | 33 | ON CONFLICT (mr_id) DO UPDATE SET user_id = excluded.user_id, |
| 25 | strategy = excluded.strategy, reason = '', | 34 | strategy = excluded.strategy, credential = excluded.credential, |
| 35 | key_id = excluded.key_id, token_id = excluded.token_id, reason = '', | ||
| 26 | queued_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')`, | 36 | queued_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')`, |
| 27 | mrID, userID, strategy) | 37 | mrID, userID, strategy, credential, nullID(keyID), nullID(tokenID)) |
| 28 | return err | 38 | return err |
| 29 | } | 39 | } |
| 30 | 40 | ||
| 41 | // QueueCredential is the credential a queued merge was queued with. Kind | ||
| 42 | // is "key", "token", or "" for a web session; an id of 0 under "key" or | ||
| 43 | // "token" means that credential has since been removed. | ||
| 44 | type QueueCredential struct { | ||
| 45 | Kind string | ||
| 46 | KeyID int64 | ||
| 47 | TokenID int64 | ||
| 48 | } | ||
| 49 | |||
| 50 | func (s *Store) MergeQueueCredential(mrID int64) (QueueCredential, error) { | ||
| 51 | var q QueueCredential | ||
| 52 | err := s.DB.QueryRow( | ||
| 53 | "SELECT credential, COALESCE(key_id, 0), COALESCE(token_id, 0) FROM mr_merge_queue WHERE mr_id = ?", | ||
| 54 | mrID).Scan(&q.Kind, &q.KeyID, &q.TokenID) | ||
| 55 | if errors.Is(err, sql.ErrNoRows) { | ||
| 56 | return q, ErrNotFound | ||
| 57 | } | ||
| 58 | return q, err | ||
| 59 | } | ||
| 60 | |||
| 31 | // DequeueMerge takes a merge request off the queue, reporting whether it | 61 | // DequeueMerge takes a merge request off the queue, reporting whether it |
| 32 | // was on it. | 62 | // was on it. |
| 33 | func (s *Store) DequeueMerge(mrID int64) (bool, error) { | 63 | func (s *Store) DequeueMerge(mrID int64) (bool, error) { |
| @@ -51,7 +81,7 @@ func (s *Store) SetMergeQueueReason(mrID int64, reason string) error { | |||
| 51 | func (s *Store) QueuedMRsAtHead(repoID int64, sha string) ([]int64, error) { | 81 | func (s *Store) QueuedMRsAtHead(repoID int64, sha string) ([]int64, error) { |
| 52 | rows, err := s.DB.Query(` | 82 | rows, err := s.DB.Query(` |
| 53 | SELECT m.id FROM mr_merge_queue q JOIN merge_requests m ON m.id = q.mr_id | 83 | SELECT m.id FROM mr_merge_queue q JOIN merge_requests m ON m.id = q.mr_id |
| 54 | WHERE m.repo_id = ? AND m.head_sha = ? AND m.state IN ('open', 'source_gone') | 84 | WHERE m.repo_id = ? AND m.head_sha = ? AND m.state = 'open' |
| 55 | ORDER BY m.number`, repoID, sha) | 85 | ORDER BY m.number`, repoID, sha) |
| 56 | if err != nil { | 86 | if err != nil { |
| 57 | return nil, err | 87 | return nil, err |
internal/store/mergequeue_test.go +49 −4
| @@ -17,7 +17,7 @@ func TestMergeQueueRoundTrip(t *testing.T) { | |||
| 17 | if mr.QueuedAt != "" { | 17 | if mr.QueuedAt != "" { |
| 18 | t.Fatalf("fresh MR is queued: %+v", mr) | 18 | t.Fatalf("fresh MR is queued: %+v", mr) |
| 19 | } | 19 | } |
| 20 | if err := s.QueueMerge(mr.ID, uid, "ff"); err != nil { | 20 | if err := s.QueueMerge(mr.ID, uid, "ff", 0, 0); err != nil { |
| 21 | t.Fatal(err) | 21 | t.Fatal(err) |
| 22 | } | 22 | } |
| 23 | if err := s.SetMergeQueueReason(mr.ID, "checks pending"); err != nil { | 23 | if err := s.SetMergeQueueReason(mr.ID, "checks pending"); err != nil { |
| @@ -32,7 +32,7 @@ func TestMergeQueueRoundTrip(t *testing.T) { | |||
| 32 | t.Fatalf("MRByID = %+v, %v", byID, err) | 32 | t.Fatalf("MRByID = %+v, %v", byID, err) |
| 33 | } | 33 | } |
| 34 | 34 | ||
| 35 | if err := s.QueueMerge(mr.ID, uid, "merge"); err != nil { | 35 | if err := s.QueueMerge(mr.ID, uid, "merge", 0, 0); err != nil { |
| 36 | t.Fatal(err) | 36 | t.Fatal(err) |
| 37 | } | 37 | } |
| 38 | mr, _ = s.MRByNumber(repoID, 1) | 38 | mr, _ = s.MRByNumber(repoID, 1) |
| @@ -68,7 +68,7 @@ func TestMergeQueueLeftOnMergeOrClose(t *testing.T) { | |||
| 68 | if err := s.SetMRState(mr.ID, "open"); err != nil { | 68 | if err := s.SetMRState(mr.ID, "open"); err != nil { |
| 69 | t.Fatal(err) | 69 | t.Fatal(err) |
| 70 | } | 70 | } |
| 71 | if err := s.QueueMerge(mr.ID, uid, ""); err != nil { | 71 | if err := s.QueueMerge(mr.ID, uid, "", 0, 0); err != nil { |
| 72 | t.Fatal(err) | 72 | t.Fatal(err) |
| 73 | } | 73 | } |
| 74 | if err := mark(); err != nil { | 74 | if err := mark(); err != nil { |
| @@ -83,9 +83,54 @@ func TestMergeQueueLeftOnMergeOrClose(t *testing.T) { | |||
| 83 | if err := s.SetMRState(mr.ID, "open"); err != nil { | 83 | if err := s.SetMRState(mr.ID, "open"); err != nil { |
| 84 | t.Fatal(err) | 84 | t.Fatal(err) |
| 85 | } | 85 | } |
| 86 | s.QueueMerge(mr.ID, uid, "") | 86 | s.QueueMerge(mr.ID, uid, "", 0, 0) |
| 87 | s.SetMRState(mr.ID, "source_gone") | 87 | s.SetMRState(mr.ID, "source_gone") |
| 88 | if got, _ := s.MRByNumber(repoID, 1); got.QueuedAt == "" { | 88 | if got, _ := s.MRByNumber(repoID, 1); got.QueuedAt == "" { |
| 89 | t.Fatal("source_gone dropped the queued merge") | 89 | t.Fatal("source_gone dropped the queued merge") |
| 90 | } | 90 | } |
| 91 | } | 91 | } |
| 92 | |||
| 93 | // A queued merge remembers the credential it was queued with, and a | ||
| 94 | // removed key or revoked token reads back as gone rather than as some | ||
| 95 | // later credential that reused the id. | ||
| 96 | func TestMergeQueueCredential(t *testing.T) { | ||
| 97 | s, repoID, uid := mrFixture(t) | ||
| 98 | mr, _ := s.MRByNumber(repoID, 1) | ||
| 99 | if err := s.AddSSHKey(uid, "SHA256:k1", "ssh-ed25519", []byte("b1"), "full", ""); err != nil { | ||
| 100 | t.Fatal(err) | ||
| 101 | } | ||
| 102 | key, err := s.SSHKeyByFingerprint("SHA256:k1") | ||
| 103 | if err != nil { | ||
| 104 | t.Fatal(err) | ||
| 105 | } | ||
| 106 | if err := s.QueueMerge(mr.ID, uid, "", key.ID, 0); err != nil { | ||
| 107 | t.Fatal(err) | ||
| 108 | } | ||
| 109 | if q, err := s.MergeQueueCredential(mr.ID); err != nil || q.Kind != "key" || q.KeyID != key.ID { | ||
| 110 | t.Fatalf("credential = %+v, %v", q, err) | ||
| 111 | } | ||
| 112 | if err := s.RemoveSSHKey(uid, "SHA256:k1"); err != nil { | ||
| 113 | t.Fatal(err) | ||
| 114 | } | ||
| 115 | if q, _ := s.MergeQueueCredential(mr.ID); q.Kind != "key" || q.KeyID != 0 { | ||
| 116 | t.Fatalf("credential after key removal = %+v, want key with id 0", q) | ||
| 117 | } | ||
| 118 | if err := s.QueueMerge(mr.ID, uid, "", 0, 0); err != nil { | ||
| 119 | t.Fatal(err) | ||
| 120 | } | ||
| 121 | if q, _ := s.MergeQueueCredential(mr.ID); q.Kind != "" { | ||
| 122 | t.Fatalf("web-queued credential = %+v", q) | ||
| 123 | } | ||
| 124 | } | ||
| 125 | |||
| 126 | // A merge request whose source branch is gone is not one a status can | ||
| 127 | // merge. | ||
| 128 | func TestQueuedMRsAtHeadSkipsSourceGone(t *testing.T) { | ||
| 129 | s, repoID, uid := mrFixture(t) | ||
| 130 | mr, _ := s.MRByNumber(repoID, 1) | ||
| 131 | s.QueueMerge(mr.ID, uid, "", 0, 0) | ||
| 132 | s.SetMRState(mr.ID, "source_gone") | ||
| 133 | if ids, _ := s.QueuedMRsAtHead(repoID, "abc123"); len(ids) != 0 { | ||
| 134 | t.Fatalf("QueuedMRsAtHead = %v, want none", ids) | ||
| 135 | } | ||
| 136 | } | ||
internal/store/migrations/0067_merge_queue.up.sql +13 −5
| @@ -2,12 +2,20 @@ | |||
| 2 | -- with strategy once its gates pass. reason is why the last attempt did | 2 | -- with strategy once its gates pass. reason is why the last attempt did |
| 3 | -- not merge. A merge request that is merged or closed leaves the queue, | 3 | -- not merge. A merge request that is merged or closed leaves the queue, |
| 4 | -- whichever path did it. | 4 | -- whichever path did it. |
| 5 | -- | ||
| 6 | -- credential is what the merge was queued with: 'key' (key_id) or | ||
| 7 | -- 'token' (token_id), or '' for a web session, which binds to the | ||
| 8 | -- account alone. A removed key or revoked token nulls its id, so the id | ||
| 9 | -- of a later credential never stands in for it. | ||
| 5 | CREATE TABLE mr_merge_queue ( | 10 | CREATE TABLE mr_merge_queue ( |
| 6 | mr_id INTEGER PRIMARY KEY REFERENCES merge_requests(id) ON DELETE CASCADE, | 11 | mr_id INTEGER PRIMARY KEY REFERENCES merge_requests(id) ON DELETE CASCADE, |
| 7 | user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, | 12 | user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, |
| 8 | strategy TEXT NOT NULL DEFAULT '', | 13 | strategy TEXT NOT NULL DEFAULT '', |
| 9 | reason TEXT NOT NULL DEFAULT '', | 14 | credential TEXT NOT NULL DEFAULT '' CHECK (credential IN ('', 'key', 'token')), |
| 10 | queued_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) | 15 | key_id INTEGER REFERENCES ssh_keys(id) ON DELETE SET NULL, |
| 16 | token_id INTEGER REFERENCES api_tokens(id) ON DELETE SET NULL, | ||
| 17 | reason TEXT NOT NULL DEFAULT '', | ||
| 18 | queued_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) | ||
| 11 | ); | 19 | ); |
| 12 | CREATE INDEX mr_merge_queue_user ON mr_merge_queue(user_id); | 20 | CREATE INDEX mr_merge_queue_user ON mr_merge_queue(user_id); |
| 13 | CREATE TRIGGER mr_merge_queue_leave AFTER UPDATE OF state ON merge_requests | 21 | CREATE TRIGGER mr_merge_queue_leave AFTER UPDATE OF state ON merge_requests |
internal/store/tokens.go +13
| @@ -64,6 +64,19 @@ func (s *Store) APITokenUser(tokenHash string) (User, APIToken, error) { | |||
| 64 | return u, t, err | 64 | return u, t, err |
| 65 | } | 65 | } |
| 66 | 66 | ||
| 67 | // APITokenByID loads a token by id, expired or not. | ||
| 68 | func (s *Store) APITokenByID(id int64) (APIToken, error) { | ||
| 69 | var t APIToken | ||
| 70 | var exp sql.NullString | ||
| 71 | err := s.DB.QueryRow("SELECT id, name, scope, created_at, expires_at FROM api_tokens WHERE id = ?", id). | ||
| 72 | Scan(&t.ID, &t.Name, &t.Scope, &t.CreatedAt, &exp) | ||
| 73 | if errors.Is(err, sql.ErrNoRows) { | ||
| 74 | return t, ErrNotFound | ||
| 75 | } | ||
| 76 | t.ExpiresAt = parseTime(exp) | ||
| 77 | return t, err | ||
| 78 | } | ||
| 79 | |||
| 67 | func (s *Store) ListAPITokens(userID int64) ([]APIToken, error) { | 80 | func (s *Store) ListAPITokens(userID int64) ([]APIToken, error) { |
| 68 | rows, err := s.DB.Query(` | 81 | rows, err := s.DB.Query(` |
| 69 | SELECT t.id, t.name, t.scope, t.created_at, t.expires_at, t.last_used_at, COALESCE(p.name, '') | 82 | SELECT t.id, t.name, t.scope, t.created_at, t.expires_at, t.last_used_at, COALESCE(p.name, '') |