Skip to content

Commit 8b33f80

Browse files
zjjhotzjjhot
committed
Merge remote-tracking branch 'upstream/main'
* upstream/main: Correctly handle select on multiple channels in Queues (go-gitea#22146) Support template for merge message description (go-gitea#22248)
2 parents 1217216 + a609cae commit 8b33f80

File tree

10 files changed

+146
-82
lines changed

10 files changed

+146
-82
lines changed

‎modules/queue/queue_channel.go

-26
Original file line numberDiff line numberDiff line change
@@ -109,32 +109,6 @@ func (q *ChannelQueue) Flush(timeout time.Duration) error {
109109
return q.FlushWithContext(ctx)
110110
}
111111

112-
// FlushWithContext is very similar to CleanUp but it will return as soon as the dataChan is empty
113-
func (q *ChannelQueue) FlushWithContext(ctx context.Context) error {
114-
log.Trace("ChannelQueue: %d Flush", q.qid)
115-
paused, _ := q.IsPausedIsResumed()
116-
for {
117-
select {
118-
case <-paused:
119-
return nil
120-
case data, ok := <-q.dataChan:
121-
if !ok {
122-
return nil
123-
}
124-
if unhandled := q.handle(data); unhandled != nil {
125-
log.Error("Unhandled Data whilst flushing queue %d", q.qid)
126-
}
127-
atomic.AddInt64(&q.numInQueue, -1)
128-
case <-q.baseCtx.Done():
129-
return q.baseCtx.Err()
130-
case <-ctx.Done():
131-
return ctx.Err()
132-
default:
133-
return nil
134-
}
135-
}
136-
}
137-
138112
// Shutdown processing from this queue
139113
func (q *ChannelQueue) Shutdown() {
140114
q.lock.Lock()

‎modules/queue/unique_queue_channel.go

-30
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import (
88
"fmt"
99
"runtime/pprof"
1010
"sync"
11-
"sync/atomic"
1211
"time"
1312

1413
"code.gitea.io/gitea/modules/container"
@@ -167,35 +166,6 @@ func (q *ChannelUniqueQueue) Flush(timeout time.Duration) error {
167166
return q.FlushWithContext(ctx)
168167
}
169168

170-
// FlushWithContext is very similar to CleanUp but it will return as soon as the dataChan is empty
171-
func (q *ChannelUniqueQueue) FlushWithContext(ctx context.Context) error {
172-
log.Trace("ChannelUniqueQueue: %d Flush", q.qid)
173-
paused, _ := q.IsPausedIsResumed()
174-
for {
175-
select {
176-
case <-paused:
177-
return nil
178-
default:
179-
}
180-
select {
181-
case data, ok := <-q.dataChan:
182-
if !ok {
183-
return nil
184-
}
185-
if unhandled := q.handle(data); unhandled != nil {
186-
log.Error("Unhandled Data whilst flushing queue %d", q.qid)
187-
}
188-
atomic.AddInt64(&q.numInQueue, -1)
189-
case <-q.baseCtx.Done():
190-
return q.baseCtx.Err()
191-
case <-ctx.Done():
192-
return ctx.Err()
193-
default:
194-
return nil
195-
}
196-
}
197-
}
198-
199169
// Shutdown processing from this queue
200170
func (q *ChannelUniqueQueue) Shutdown() {
201171
log.Trace("ChannelUniqueQueue: %s Shutting down", q.name)

‎modules/queue/workerpool.go

+43-1
Original file line numberDiff line numberDiff line change
@@ -463,13 +463,43 @@ func (p *WorkerPool) IsEmpty() bool {
463463
return atomic.LoadInt64(&p.numInQueue) == 0
464464
}
465465

466+
// contextError returns either ctx.Done(), the base context's error or nil
467+
func (p *WorkerPool) contextError(ctx context.Context) error {
468+
select {
469+
case <-p.baseCtx.Done():
470+
return p.baseCtx.Err()
471+
case <-ctx.Done():
472+
return ctx.Err()
473+
default:
474+
return nil
475+
}
476+
}
477+
466478
// FlushWithContext is very similar to CleanUp but it will return as soon as the dataChan is empty
467479
// NB: The worker will not be registered with the manager.
468480
func (p *WorkerPool) FlushWithContext(ctx context.Context) error {
469481
log.Trace("WorkerPool: %d Flush", p.qid)
482+
paused, _ := p.IsPausedIsResumed()
470483
for {
484+
// Because select will return any case that is satisified at random we precheck here before looking at dataChan.
485+
select {
486+
case <-paused:
487+
// Ensure that even if paused that the cancelled error is still sent
488+
return p.contextError(ctx)
489+
case <-p.baseCtx.Done():
490+
return p.baseCtx.Err()
491+
case <-ctx.Done():
492+
return ctx.Err()
493+
default:
494+
}
495+
471496
select {
472-
case data := <-p.dataChan:
497+
case <-paused:
498+
return p.contextError(ctx)
499+
case data, ok := <-p.dataChan:
500+
if !ok {
501+
return nil
502+
}
473503
if unhandled := p.handle(data); unhandled != nil {
474504
log.Error("Unhandled Data whilst flushing queue %d", p.qid)
475505
}
@@ -495,6 +525,7 @@ func (p *WorkerPool) doWork(ctx context.Context) {
495525
paused, _ := p.IsPausedIsResumed()
496526
data := make([]Data, 0, p.batchLength)
497527
for {
528+
// Because select will return any case that is satisified at random we precheck here before looking at dataChan.
498529
select {
499530
case <-paused:
500531
log.Trace("Worker for Queue %d Pausing", p.qid)
@@ -515,8 +546,19 @@ func (p *WorkerPool) doWork(ctx context.Context) {
515546
log.Trace("Worker shutting down")
516547
return
517548
}
549+
case <-ctx.Done():
550+
if len(data) > 0 {
551+
log.Trace("Handling: %d data, %v", len(data), data)
552+
if unhandled := p.handle(data...); unhandled != nil {
553+
log.Error("Unhandled Data in queue %d", p.qid)
554+
}
555+
atomic.AddInt64(&p.numInQueue, -1*int64(len(data)))
556+
}
557+
log.Trace("Worker shutting down")
558+
return
518559
default:
519560
}
561+
520562
select {
521563
case <-paused:
522564
// go back around

‎routers/api/v1/repo/pull.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -815,7 +815,7 @@ func MergePullRequest(ctx *context.APIContext) {
815815

816816
message := strings.TrimSpace(form.MergeTitleField)
817817
if len(message) == 0 {
818-
message, err = pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pr, repo_model.MergeStyle(form.Do))
818+
message, _, err = pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pr, repo_model.MergeStyle(form.Do))
819819
if err != nil {
820820
ctx.Error(http.StatusInternalServerError, "GetDefaultMergeMessage", err)
821821
return

‎routers/web/repo/issue.go

+4-2
Original file line numberDiff line numberDiff line change
@@ -1664,19 +1664,21 @@ func ViewIssue(ctx *context.Context) {
16641664

16651665
ctx.Data["MergeStyle"] = mergeStyle
16661666

1667-
defaultMergeMessage, err := pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pull, mergeStyle)
1667+
defaultMergeMessage, defaultMergeBody, err := pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pull, mergeStyle)
16681668
if err != nil {
16691669
ctx.ServerError("GetDefaultMergeMessage", err)
16701670
return
16711671
}
16721672
ctx.Data["DefaultMergeMessage"] = defaultMergeMessage
1673+
ctx.Data["DefaultMergeBody"] = defaultMergeBody
16731674

1674-
defaultSquashMergeMessage, err := pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pull, repo_model.MergeStyleSquash)
1675+
defaultSquashMergeMessage, defaultSquashMergeBody, err := pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pull, repo_model.MergeStyleSquash)
16751676
if err != nil {
16761677
ctx.ServerError("GetDefaultSquashMergeMessage", err)
16771678
return
16781679
}
16791680
ctx.Data["DefaultSquashMergeMessage"] = defaultSquashMergeMessage
1681+
ctx.Data["DefaultSquashMergeBody"] = defaultSquashMergeBody
16801682

16811683
if err = pull.LoadProtectedBranch(ctx); err != nil {
16821684
ctx.ServerError("LoadProtectedBranch", err)

‎routers/web/repo/pull.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -986,7 +986,7 @@ func MergePullRequest(ctx *context.Context) {
986986
message := strings.TrimSpace(form.MergeTitleField)
987987
if len(message) == 0 {
988988
var err error
989-
message, err = pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pr, repo_model.MergeStyle(form.Do))
989+
message, _, err = pull_service.GetDefaultMergeMessage(ctx, ctx.Repo.GitRepo, pr, repo_model.MergeStyle(form.Do))
990990
if err != nil {
991991
ctx.ServerError("GetDefaultMergeMessage", err)
992992
return

‎services/pull/merge.go

+23-15
Original file line numberDiff line numberDiff line change
@@ -39,19 +39,19 @@ import (
3939
)
4040

4141
// GetDefaultMergeMessage returns default message used when merging pull request
42-
func GetDefaultMergeMessage(ctx context.Context, baseGitRepo *git.Repository, pr *issues_model.PullRequest, mergeStyle repo_model.MergeStyle) (string, error) {
42+
func GetDefaultMergeMessage(ctx context.Context, baseGitRepo *git.Repository, pr *issues_model.PullRequest, mergeStyle repo_model.MergeStyle) (message, body string, err error) {
4343
if err := pr.LoadHeadRepo(ctx); err != nil {
44-
return "", err
44+
return "", "", err
4545
}
4646
if err := pr.LoadBaseRepo(ctx); err != nil {
47-
return "", err
47+
return "", "", err
4848
}
4949
if pr.BaseRepo == nil {
50-
return "", repo_model.ErrRepoNotExist{ID: pr.BaseRepoID}
50+
return "", "", repo_model.ErrRepoNotExist{ID: pr.BaseRepoID}
5151
}
5252

5353
if err := pr.LoadIssue(ctx); err != nil {
54-
return "", err
54+
return "", "", err
5555
}
5656

5757
isExternalTracker := pr.BaseRepo.UnitEnabled(ctx, unit.TypeExternalTracker)
@@ -64,12 +64,12 @@ func GetDefaultMergeMessage(ctx context.Context, baseGitRepo *git.Repository, pr
6464
templateFilepath := fmt.Sprintf(".gitea/default_merge_message/%s_TEMPLATE.md", strings.ToUpper(string(mergeStyle)))
6565
commit, err := baseGitRepo.GetBranchCommit(pr.BaseRepo.DefaultBranch)
6666
if err != nil {
67-
return "", err
67+
return "", "", err
6868
}
6969
templateContent, err := commit.GetFileContent(templateFilepath, setting.Repository.PullRequest.DefaultMergeMessageSize)
7070
if err != nil {
7171
if !git.IsErrNotExist(err) {
72-
return "", err
72+
return "", "", err
7373
}
7474
} else {
7575
vars := map[string]string{
@@ -107,27 +107,35 @@ func GetDefaultMergeMessage(ctx context.Context, baseGitRepo *git.Repository, pr
107107
vars["ClosingIssues"] = ""
108108
}
109109
}
110-
111-
return os.Expand(templateContent, func(s string) string {
112-
return vars[s]
113-
}), nil
110+
message, body = expandDefaultMergeMessage(templateContent, vars)
111+
return message, body, nil
114112
}
115113
}
116114

117115
// Squash merge has a different from other styles.
118116
if mergeStyle == repo_model.MergeStyleSquash {
119-
return fmt.Sprintf("%s (%s%d)", pr.Issue.Title, issueReference, pr.Issue.Index), nil
117+
return fmt.Sprintf("%s (%s%d)", pr.Issue.Title, issueReference, pr.Issue.Index), "", nil
120118
}
121119

122120
if pr.BaseRepoID == pr.HeadRepoID {
123-
return fmt.Sprintf("Merge pull request '%s' (%s%d) from %s into %s", pr.Issue.Title, issueReference, pr.Issue.Index, pr.HeadBranch, pr.BaseBranch), nil
121+
return fmt.Sprintf("Merge pull request '%s' (%s%d) from %s into %s", pr.Issue.Title, issueReference, pr.Issue.Index, pr.HeadBranch, pr.BaseBranch), "", nil
124122
}
125123

126124
if pr.HeadRepo == nil {
127-
return fmt.Sprintf("Merge pull request '%s' (%s%d) from <deleted>:%s into %s", pr.Issue.Title, issueReference, pr.Issue.Index, pr.HeadBranch, pr.BaseBranch), nil
125+
return fmt.Sprintf("Merge pull request '%s' (%s%d) from <deleted>:%s into %s", pr.Issue.Title, issueReference, pr.Issue.Index, pr.HeadBranch, pr.BaseBranch), "", nil
128126
}
129127

130-
return fmt.Sprintf("Merge pull request '%s' (%s%d) from %s:%s into %s", pr.Issue.Title, issueReference, pr.Issue.Index, pr.HeadRepo.FullName(), pr.HeadBranch, pr.BaseBranch), nil
128+
return fmt.Sprintf("Merge pull request '%s' (%s%d) from %s:%s into %s", pr.Issue.Title, issueReference, pr.Issue.Index, pr.HeadRepo.FullName(), pr.HeadBranch, pr.BaseBranch), "", nil
129+
}
130+
131+
func expandDefaultMergeMessage(template string, vars map[string]string) (message, body string) {
132+
message = strings.TrimSpace(template)
133+
if splits := strings.SplitN(message, "\n", 2); len(splits) == 2 {
134+
message = splits[0]
135+
body = strings.TrimSpace(splits[1])
136+
}
137+
mapping := func(s string) string { return vars[s] }
138+
return os.Expand(message, mapping), os.Expand(body, mapping)
131139
}
132140

133141
// Merge merges pull request to base repository.

‎services/pull/merge_test.go

+67
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
// Copyright 2022 The Gitea Authors. All rights reserved.
2+
// SPDX-License-Identifier: MIT
3+
4+
package pull
5+
6+
import (
7+
"testing"
8+
9+
"github.com/stretchr/testify/assert"
10+
)
11+
12+
func Test_expandDefaultMergeMessage(t *testing.T) {
13+
type args struct {
14+
template string
15+
vars map[string]string
16+
}
17+
tests := []struct {
18+
name string
19+
args args
20+
want string
21+
wantBody string
22+
}{
23+
{
24+
name: "single line",
25+
args: args{
26+
template: "Merge ${PullRequestTitle}",
27+
vars: map[string]string{
28+
"PullRequestTitle": "PullRequestTitle",
29+
"PullRequestDescription": "Pull\nRequest\nDescription\n",
30+
},
31+
},
32+
want: "Merge PullRequestTitle",
33+
wantBody: "",
34+
},
35+
{
36+
name: "multiple lines",
37+
args: args{
38+
template: "Merge ${PullRequestTitle}\nDescription:\n\n${PullRequestDescription}\n",
39+
vars: map[string]string{
40+
"PullRequestTitle": "PullRequestTitle",
41+
"PullRequestDescription": "Pull\nRequest\nDescription\n",
42+
},
43+
},
44+
want: "Merge PullRequestTitle",
45+
wantBody: "Description:\n\nPull\nRequest\nDescription\n",
46+
},
47+
{
48+
name: "leading newlines",
49+
args: args{
50+
template: "\n\n\nMerge ${PullRequestTitle}\n\n\nDescription:\n\n${PullRequestDescription}\n",
51+
vars: map[string]string{
52+
"PullRequestTitle": "PullRequestTitle",
53+
"PullRequestDescription": "Pull\nRequest\nDescription\n",
54+
},
55+
},
56+
want: "Merge PullRequestTitle",
57+
wantBody: "Description:\n\nPull\nRequest\nDescription\n",
58+
},
59+
}
60+
for _, tt := range tests {
61+
t.Run(tt.name, func(t *testing.T) {
62+
got, got1 := expandDefaultMergeMessage(tt.args.template, tt.args.vars)
63+
assert.Equalf(t, tt.want, got, "expandDefaultMergeMessage(%v, %v)", tt.args.template, tt.args.vars)
64+
assert.Equalf(t, tt.wantBody, got1, "expandDefaultMergeMessage(%v, %v)", tt.args.template, tt.args.vars)
65+
})
66+
}
67+
}

‎services/pull/pull_test.go

+4-4
Original file line numberDiff line numberDiff line change
@@ -45,13 +45,13 @@ func TestPullRequest_GetDefaultMergeMessage_InternalTracker(t *testing.T) {
4545
assert.NoError(t, err)
4646
defer gitRepo.Close()
4747

48-
mergeMessage, err := GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
48+
mergeMessage, _, err := GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
4949
assert.NoError(t, err)
5050
assert.Equal(t, "Merge pull request 'issue3' (#3) from branch2 into master", mergeMessage)
5151

5252
pr.BaseRepoID = 1
5353
pr.HeadRepoID = 2
54-
mergeMessage, err = GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
54+
mergeMessage, _, err = GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
5555
assert.NoError(t, err)
5656
assert.Equal(t, "Merge pull request 'issue3' (#3) from user2/repo1:branch2 into master", mergeMessage)
5757
}
@@ -75,7 +75,7 @@ func TestPullRequest_GetDefaultMergeMessage_ExternalTracker(t *testing.T) {
7575
assert.NoError(t, err)
7676
defer gitRepo.Close()
7777

78-
mergeMessage, err := GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
78+
mergeMessage, _, err := GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
7979
assert.NoError(t, err)
8080

8181
assert.Equal(t, "Merge pull request 'issue3' (!3) from branch2 into master", mergeMessage)
@@ -84,7 +84,7 @@ func TestPullRequest_GetDefaultMergeMessage_ExternalTracker(t *testing.T) {
8484
pr.HeadRepoID = 2
8585
pr.BaseRepo = nil
8686
pr.HeadRepo = nil
87-
mergeMessage, err = GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
87+
mergeMessage, _, err = GetDefaultMergeMessage(db.DefaultContext, gitRepo, pr, "")
8888
assert.NoError(t, err)
8989

9090
assert.Equal(t, "Merge pull request 'issue3' (#3) from user2/repo2:branch2 into master", mergeMessage)

‎templates/repo/issue/view_content/pull.tmpl

+3-2
Original file line numberDiff line numberDiff line change
@@ -343,7 +343,8 @@
343343
(() => {
344344
const defaultMergeTitle = {{.DefaultMergeMessage}};
345345
const defaultSquashMergeTitle = {{.DefaultSquashMergeMessage}};
346-
const defaultMergeMessage = 'Reviewed-on: ' + {{$.Issue.HTMLURL}} + '\n' + {{$approvers}};
346+
const defaultMergeMessage = {{if .DefaultMergeBody}}{{.DefaultMergeBody}}{{else}}'Reviewed-on: ' + {{$.Issue.HTMLURL}} + '\n' + {{$approvers}}{{end}};
347+
const defaultSquashMergeMessage = {{if .DefaultSquashMergeBody}}{{.DefaultSquashMergeBody}}{{else}}'Reviewed-on: ' + {{$.Issue.HTMLURL}} + '\n' + {{$approvers}}{{end}};
347348
const mergeForm = {
348349
'baseLink': {{.Link}},
349350
'textCancel': {{$.locale.Tr "cancel"}},
@@ -398,7 +399,7 @@
398399
'allowed': {{$prUnit.PullRequestsConfig.AllowSquash}},
399400
'textDoMerge': {{$.locale.Tr "repo.pulls.squash_merge_pull_request"}},
400401
'mergeTitleFieldText': defaultSquashMergeTitle,
401-
'mergeMessageFieldText': {{.GetCommitMessages}} + defaultMergeMessage,
402+
'mergeMessageFieldText': {{.GetCommitMessages}} + defaultSquashMergeMessage,
402403
'hideAutoMerge': generalHideAutoMerge,
403404
},
404405
{

0 commit comments

Comments
 (0)