fix(usecase): verify the proof of work before any admission check and reserve quota and queue atomically

commit: b5344bf3cd0d9d3ebc96273a987239a1ad334918
author: brice · 3d ago · 2026-10-19 08:25 UTC
parent: bc78630
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +413 -151

M docs/issues.md +1 -1


@@ -44,7 +44,7 @@ The page updates live: merge request and issue events of any repository refresh
44	44
45	45		With `issues` set to `public` on a public repository, a visitor who is not signed in finds a New issue button on the Issues tab, which opens `/<repo>/issues/new`. The form needs JavaScript: the page solves a short proof of work in a worker while the visitor writes (a second or so; harder after many recent submissions), and the button stays disabled until it is done and a few seconds have passed. Without JavaScript the form stays hidden and the page says why. A repository that is private, or whose mode is `off` or `owner`, has no such page: it answers the same 404 as an unknown one.
46	46
47		-	Limits: a title of at most 120 characters, a description of at most 10,000, an optional name of at most 40 (not verified). Each network may submit 3 issues per hour and 10 per day. At most 200 submissions await moderation; beyond that the form says submissions are paused. A submission never appears on its own: it is stored as pending and shows no id to the visitor.
	47	+	Limits: a title of at most 120 characters, a description of at most 10,000, an optional name of at most 40 (not verified). Each network may submit 3 issues per hour and 10 per day. At most 200 submissions await moderation; beyond that the form says submissions are paused. Before the proof of work is valid, nothing reveals whether a network is banned or limited, or whether the queue is full: every such failure gets the same generic refusal. A submission never appears on its own: it is stored as pending and shows no id to the visitor.
48	48
49	49		The owner moderates in the Inbox, in the first section, Awaiting moderation, whose count is in the counters row. Each submission shows its repository, age, name (or "anonymous"), title and the start of the description, as plain text, with three buttons: Approve opens it as the next issue of the repository (numbers are shared with merge requests), Reject deletes it, Reject and ban deletes it and refuses further submissions from the same network for 30 days. The section updates live when a submission arrives.
50	50


@@ -44,7 +44,7 @@ The page updates live: merge request and issue events of any repository refresh
44			44
45		With `issues` set to `public` on a public repository, a visitor who is not signed in finds a New issue button on the Issues tab, which opens `/<repo>/issues/new`. The form needs JavaScript: the page solves a short proof of work in a worker while the visitor writes (a second or so; harder after many recent submissions), and the button stays disabled until it is done and a few seconds have passed. Without JavaScript the form stays hidden and the page says why. A repository that is private, or whose mode is `off` or `owner`, has no such page: it answers the same 404 as an unknown one.	45		With `issues` set to `public` on a public repository, a visitor who is not signed in finds a New issue button on the Issues tab, which opens `/<repo>/issues/new`. The form needs JavaScript: the page solves a short proof of work in a worker while the visitor writes (a second or so; harder after many recent submissions), and the button stays disabled until it is done and a few seconds have passed. Without JavaScript the form stays hidden and the page says why. A repository that is private, or whose mode is `off` or `owner`, has no such page: it answers the same 404 as an unknown one.
46			46
47	-	Limits: a title of at most 120 characters, a description of at most 10,000, an optional name of at most 40 (not verified). Each network may submit 3 issues per hour and 10 per day. At most 200 submissions await moderation; beyond that the form says submissions are paused. A submission never appears on its own: it is stored as pending and shows no id to the visitor.	47	+	Limits: a title of at most 120 characters, a description of at most 10,000, an optional name of at most 40 (not verified). Each network may submit 3 issues per hour and 10 per day. At most 200 submissions await moderation; beyond that the form says submissions are paused. Before the proof of work is valid, nothing reveals whether a network is banned or limited, or whether the queue is full: every such failure gets the same generic refusal. A submission never appears on its own: it is stored as pending and shows no id to the visitor.
48			48
49		The owner moderates in the Inbox, in the first section, Awaiting moderation, whose count is in the counters row. Each submission shows its repository, age, name (or "anonymous"), title and the start of the description, as plain text, with three buttons: Approve opens it as the next issue of the repository (numbers are shared with merge requests), Reject deletes it, Reject and ban deletes it and refuses further submissions from the same network for 30 days. The section updates live when a submission arrives.	49		The owner moderates in the Inbox, in the first section, Awaiting moderation, whose count is in the counters row. Each submission shows its repository, age, name (or "anonymous"), title and the start of the description, as plain text, with three buttons: Approve opens it as the next issue of the repository (numbers are shared with merge requests), Reject deletes it, Reject and ban deletes it and refuses further submissions from the same network for 30 days. The section updates live when a submission arrives.
50			50

M internal/adapters/web/issues_form_test.go +22 -0


@@ -178,6 +178,28 @@ func TestIssueFormPostOutcomes(t *testing.T) {
178	178		}
179	179		}
180	180
	181	+	// TestIssueFormPostInvalidTokenLooksTheSame: the use case answers an invalid
	182	+	// submission the same whatever the client's ban, limit or the queue; the page
	183	+	// must not differ either.
	184	+	func TestIssueFormPostInvalidTokenLooksTheSame(t *testing.T) {
	185	+	t.Parallel()
	186	+	var first *httptest.ResponseRecorder
	187	+	for _, state := range []string{"banned", "rate limited", "queue full"} {
	188	+	r := newFormRig(t, domain.IssuesPublic)
	189	+	r.form.EXPECT().Submit(mock.Anything, repoName(t, "proj"), mock.Anything, mock.Anything).Return(domain.ErrSubmissionInvalid).Once()
	190	+	f := goodForm()
	191	+	f.Set("token", "garbage")
	192	+	rec := r.post("/proj/issues/new", fromSite, f.Encode())
	193	+	assert.Equal(t, 400, rec.Code, state)
	194	+	if first == nil {
	195	+	first = rec
	196	+	continue
	197	+	}
	198	+	assert.Equal(t, first.Body.String(), rec.Body.String(), state)
	199	+	assert.Equal(t, first.Header().Get("Retry-After"), rec.Header().Get("Retry-After"), state)
	200	+	}
	201	+	}
	202	+
181	203		func TestIssueFormPostSuccess(t *testing.T) {
182	204		t.Parallel()
183	205		r := newFormRig(t, domain.IssuesPublic)


@@ -178,6 +178,28 @@ func TestIssueFormPostOutcomes(t *testing.T) {
178	}	178		}
179	}	179		}
180		180
		181	+	// TestIssueFormPostInvalidTokenLooksTheSame: the use case answers an invalid
		182	+	// submission the same whatever the client's ban, limit or the queue; the page
		183	+	// must not differ either.
		184	+	func TestIssueFormPostInvalidTokenLooksTheSame(t *testing.T) {
		185	+	t.Parallel()
		186	+	var first *httptest.ResponseRecorder
		187	+	for _, state := range []string{"banned", "rate limited", "queue full"} {
		188	+	r := newFormRig(t, domain.IssuesPublic)
		189	+	r.form.EXPECT().Submit(mock.Anything, repoName(t, "proj"), mock.Anything, mock.Anything).Return(domain.ErrSubmissionInvalid).Once()
		190	+	f := goodForm()
		191	+	f.Set("token", "garbage")
		192	+	rec := r.post("/proj/issues/new", fromSite, f.Encode())
		193	+	assert.Equal(t, 400, rec.Code, state)
		194	+	if first == nil {
		195	+	first = rec
		196	+	continue
		197	+	}
		198	+	assert.Equal(t, first.Body.String(), rec.Body.String(), state)
		199	+	assert.Equal(t, first.Header().Get("Retry-After"), rec.Header().Get("Retry-After"), state)
		200	+	}
		201	+	}
		202	+
181	func TestIssueFormPostSuccess(t *testing.T) {	203		func TestIssueFormPostSuccess(t *testing.T) {
182	t.Parallel()	204		t.Parallel()
183	r := newFormRig(t, domain.IssuesPublic)	205		r := newFormRig(t, domain.IssuesPublic)

M internal/clientlimit/submissions.go +35 -16


@@ -52,13 +52,17 @@ func NewSubmissionLimiter(now func() time.Time) *SubmissionLimiter {
52	52		return &SubmissionLimiter{now: now, subs: make(map[string][]time.Time)}
53	53		}
54	54
55		-	// Allow reports whether addr may submit now. When it may not, retryAfter is
56		-	// how long until the oldest submission that counts leaves its window.
57		-	func (l *SubmissionLimiter) Allow(_ context.Context, addr string) (bool, time.Duration) {
	55	+	// Reserve checks the limits for addr and, if it may submit now, counts a
	56	+	// submission, atomically: concurrent callers cannot all pass the check. When
	57	+	// addr may not submit, retryAfter is how long until the oldest submission that
	58	+	// counts leaves its window and release is a no-op. Otherwise release refunds
	59	+	// the reservation; call it only if the submission was not stored.
	60	+	func (l *SubmissionLimiter) Reserve(_ context.Context, addr string) (ok bool, retryAfter time.Duration, release func()) {
58	61		key := limiterKey(addr)
59	62		now := l.now()
60	63		l.mu.Lock()
61	64		defer l.mu.Unlock()
	65	+	l.maintain(now)
62	66		ts := l.prune(key, now)
63	67		var wait time.Duration
64	68		if h := withinWindow(ts, now, HourWindow); len(h) >= PerHour {

@@ -67,20 +71,10 @@ func (l *SubmissionLimiter) Allow(_ context.Context, addr string) (bool, time.Du
67	71		if len(ts) >= PerDay {
68	72		wait = max(wait, ts[len(ts)-PerDay].Add(DayWindow).Sub(now))
69	73		}
70		-	if wait <= 0 {
71		-	return true, 0
	74	+	if wait > 0 {
	75	+	return false, max(wait, time.Second), func() {}
72	76		}
73		-	return false, max(wait, time.Second)
74		-	}
75		-
76		-	// Record counts a submission from addr.
77		-	func (l *SubmissionLimiter) Record(_ context.Context, addr string) {
78		-	key := limiterKey(addr)
79		-	now := l.now()
80		-	l.mu.Lock()
81		-	defer l.mu.Unlock()
82		-	l.maintain(now)
83		-	ts := append(l.prune(key, now), now)
	77	+	ts = append(ts, now)
84	78		if len(ts) > PerDay*2 {
85	79		ts = ts[len(ts)-PerDay*2:] // enough to compute the wait
86	80		}

@@ -89,6 +83,31 @@ func (l *SubmissionLimiter) Record(_ context.Context, addr string) {
89	83		if len(l.global) > maxTracked {
90	84		l.global = l.global[len(l.global)-maxTracked:] // bound the memory
91	85		}
	86	+	var once sync.Once
	87	+	return true, 0, func() { once.Do(func() { l.refund(key, now) }) }
	88	+	}
	89	+
	90	+	// refund removes one reservation made at t for key.
	91	+	func (l *SubmissionLimiter) refund(key string, t time.Time) {
	92	+	l.mu.Lock()
	93	+	defer l.mu.Unlock()
	94	+	if ts := l.subs[key]; ts != nil {
	95	+	l.subs[key] = removeLast(ts, t)
	96	+	if len(l.subs[key]) == 0 {
	97	+	delete(l.subs, key)
	98	+	}
	99	+	}
	100	+	l.global = removeLast(l.global, t)
	101	+	}
	102	+
	103	+	// removeLast removes the last element of ts equal to t, if any.
	104	+	func removeLast(ts []time.Time, t time.Time) []time.Time {
	105	+	for i := len(ts) - 1; i >= 0; i-- {
	106	+	if ts[i].Equal(t) {
	107	+	return slices.Delete(ts, i, i+1)
	108	+	}
	109	+	}
	110	+	return ts
92	111		}
93	112
94	113		// Recent returns the submissions of addr and of all clients in the last hour.

 	return &SubmissionLimiter{now: now, subs: make(map[string][]time.Time)}
 }
-// Allow reports whether addr may submit now. When it may not, retryAfter is
+// Reserve checks the limits for addr and, if it may submit now, counts a
-// how long until the oldest submission that counts leaves its window.
+// submission, atomically: concurrent callers cannot all pass the check. When
-func (l *SubmissionLimiter) Allow(_ context.Context, addr string) (bool, time.Duration) {
+// addr may not submit, retryAfter is how long until the oldest submission that
+// counts leaves its window and release is a no-op. Otherwise release refunds
+// the reservation; call it only if the submission was not stored.
+func (l *SubmissionLimiter) Reserve(_ context.Context, addr string) (ok bool, retryAfter time.Duration, release func()) {
 	key := limiterKey(addr)
 	now := l.now()
 	l.mu.Lock()
 	defer l.mu.Unlock()
+	l.maintain(now)
 	ts := l.prune(key, now)
 	var wait time.Duration
 	if h := withinWindow(ts, now, HourWindow); len(h) >= PerHour {
 	if len(ts) >= PerDay {
 		wait = max(wait, ts[len(ts)-PerDay].Add(DayWindow).Sub(now))
 	}
-	if wait <= 0 {
+	if wait > 0 {
-		return true, 0
+		return false, max(wait, time.Second), func() {}
 	}
-	return false, max(wait, time.Second)
+	ts = append(ts, now)
-}
-// Record counts a submission from addr.
-func (l *SubmissionLimiter) Record(_ context.Context, addr string) {
-	key := limiterKey(addr)
-	now := l.now()
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	l.maintain(now)
-	ts := append(l.prune(key, now), now)
 	if len(ts) > PerDay*2 {
 		ts = ts[len(ts)-PerDay*2:] // enough to compute the wait
 	}
 	if len(l.global) > maxTracked {
 		l.global = l.global[len(l.global)-maxTracked:] // bound the memory
 	}
+	var once sync.Once
+	return true, 0, func() { once.Do(func() { l.refund(key, now) }) }
+}
+// refund removes one reservation made at t for key.
+func (l *SubmissionLimiter) refund(key string, t time.Time) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if ts := l.subs[key]; ts != nil {
+		l.subs[key] = removeLast(ts, t)
+		if len(l.subs[key]) == 0 {
+			delete(l.subs, key)
+		}
+	}
+	l.global = removeLast(l.global, t)
+}
+// removeLast removes the last element of ts equal to t, if any.
+func removeLast(ts []time.Time, t time.Time) []time.Time {
+	for i := len(ts) - 1; i >= 0; i-- {
+		if ts[i].Equal(t) {
+			return slices.Delete(ts, i, i+1)
+		}
+	}
+	return ts
 }
 // Recent returns the submissions of addr and of all clients in the last hour.

M internal/clientlimit/submissions_test.go +61 -19


@@ -4,6 +4,8 @@ package clientlimit_test
4	4
5	5		import (
6	6		"fmt"
	7	+	"sync"
	8	+	"sync/atomic"
7	9		"testing"
8	10		"time"
9	11

@@ -22,20 +24,19 @@ func TestSubmissionHourWindow(t *testing.T) {
22	24		t.Parallel()
23	25		l, clk := newSubs()
24	26		for i := range clientlimit.PerHour {
25		-	ok, _ := l.Allow(bg, "1.2.3.4")
	27	+	ok, _, _ := l.Reserve(bg, "1.2.3.4")
26	28		require.True(t, ok, "submission %d", i)
27		-	l.Record(bg, "1.2.3.4")
28	29		clk.t = clk.t.Add(time.Minute)
29	30		}
30		-	ok, wait := l.Allow(bg, "1.2.3.4")
	31	+	ok, wait, _ := l.Reserve(bg, "1.2.3.4")
31	32		assert.False(t, ok)
32	33		assert.Equal(t, time.Hour-3*time.Minute, wait, "until the first one leaves the hour")
33	34
34		-	other, _ := l.Allow(bg, "5.6.7.8")
	35	+	other, _, _ := l.Reserve(bg, "5.6.7.8")
35	36		assert.True(t, other, "another client is not affected")
36	37
37	38		clk.t = clk.t.Add(wait)
38		-	ok, _ = l.Allow(bg, "1.2.3.4")
	39	+	ok, _, _ = l.Reserve(bg, "1.2.3.4")
39	40		assert.True(t, ok, "the window slides")
40	41		}
41	42

@@ -45,18 +46,17 @@ func TestSubmissionDayWindow(t *testing.T) {
45	46		start := clk.t
46	47		// 3 per hour, spaced so the hourly limit never bites: one every 2 hours.
47	48		for range clientlimit.PerDay {
48		-	ok, _ := l.Allow(bg, "1.2.3.4")
	49	+	ok, _, _ := l.Reserve(bg, "1.2.3.4")
49	50		require.True(t, ok)
50		-	l.Record(bg, "1.2.3.4")
51	51		clk.t = clk.t.Add(2 * time.Hour)
52	52		}
53	53		// 20h after start: hourly window empty, day window full.
54		-	ok, wait := l.Allow(bg, "1.2.3.4")
	54	+	ok, wait, _ := l.Reserve(bg, "1.2.3.4")
55	55		assert.False(t, ok)
56	56		assert.Equal(t, start.Add(clientlimit.DayWindow).Sub(clk.t), wait)
57	57
58	58		clk.t = start.Add(clientlimit.DayWindow)
59		-	ok, _ = l.Allow(bg, "1.2.3.4")
	59	+	ok, _, _ = l.Reserve(bg, "1.2.3.4")
60	60		assert.True(t, ok)
61	61		}
62	62

@@ -64,15 +64,14 @@ func TestSubmissionIPv6GroupsBy64(t *testing.T) {
64	64		t.Parallel()
65	65		l, _ := newSubs()
66	66		for i := range clientlimit.PerHour {
67		-	l.Record(bg, fmt.Sprintf("2001:db8:1:2::%x", i+1)) // same /64
	67	+	l.Reserve(bg, fmt.Sprintf("2001:db8:1:2::%x", i+1)) // same /64
68	68		}
69		-	ok, _ := l.Allow(bg, "2001:db8:1:2:ffff::1")
	69	+	ok, _, _ := l.Reserve(bg, "2001:db8:1:2:ffff::1")
70	70		assert.False(t, ok, "the whole /64 shares the budget")
71		-	ok, _ = l.Allow(bg, "2001:db8:1:3::1")
	71	+	ok, _, _ = l.Reserve(bg, "2001:db8:1:3::1")
72	72		assert.True(t, ok, "another /64 does not")
73		-	ok, _ = l.Allow(bg, "::ffff:1.2.3.4")
	73	+	ok, _, _ = l.Reserve(bg, "::ffff:1.2.3.4")
74	74		assert.True(t, ok)
75		-	l.Record(bg, "::ffff:1.2.3.4")
76	75		perAddr, _ := l.Recent(bg, "1.2.3.4")
77	76		assert.Equal(t, 1, perAddr, "a mapped address counts as its IPv4")
78	77		}

@@ -80,9 +79,9 @@ func TestSubmissionIPv6GroupsBy64(t *testing.T) {
80	79		func TestSubmissionRecent(t *testing.T) {
81	80		t.Parallel()
82	81		l, clk := newSubs()
83		-	l.Record(bg, "1.1.1.1")
84		-	l.Record(bg, "1.1.1.1")
85		-	l.Record(bg, "2.2.2.2")
	82	+	l.Reserve(bg, "1.1.1.1")
	83	+	l.Reserve(bg, "1.1.1.1")
	84	+	l.Reserve(bg, "2.2.2.2")
86	85		perAddr, global := l.Recent(bg, "1.1.1.1")
87	86		assert.Equal(t, 2, perAddr)
88	87		assert.Equal(t, 3, global)

@@ -98,9 +97,52 @@ func TestSubmissionEvictsOldest(t *testing.T) {
98	97		l, clk := newSubs()
99	98		for i := range clientlimit.MaxTracked + 10 {
100	99		clk.t = clk.t.Add(time.Millisecond)
101		-	l.Record(bg, fmt.Sprintf("10.%d.%d.%d", i>>16&255, i>>8&255, i&255))
	100	+	l.Reserve(bg, fmt.Sprintf("10.%d.%d.%d", i>>16&255, i>>8&255, i&255))
102	101		}
103	102		assert.LessOrEqual(t, l.Tracked(), clientlimit.MaxTracked, "bounded")
104		-	ok, _ := l.Allow(bg, "10.0.0.0")
	103	+	ok, _, _ := l.Reserve(bg, "10.0.0.0")
105	104		assert.True(t, ok, "the oldest were evicted")
106	105		}
	106	+
	107	+	func TestSubmissionReserveIsAtomic(t *testing.T) {
	108	+	t.Parallel()
	109	+	l, _ := newSubs()
	110	+	const n = 50
	111	+	var granted atomic.Int32
	112	+	var wg sync.WaitGroup
	113	+	start := make(chan struct{})
	114	+	for range n {
	115	+	wg.Add(1)
	116	+	go func() {
	117	+	defer wg.Done()
	118	+	<-start
	119	+	if ok, _, _ := l.Reserve(bg, "1.2.3.4"); ok {
	120	+	granted.Add(1)
	121	+	}
	122	+	}()
	123	+	}
	124	+	close(start)
	125	+	wg.Wait()
	126	+	assert.Equal(t, int32(clientlimit.PerHour), granted.Load())
	127	+	}
	128	+
	129	+	func TestSubmissionReleaseRefunds(t *testing.T) {
	130	+	t.Parallel()
	131	+	l, _ := newSubs()
	132	+	var releases []func()
	133	+	for range clientlimit.PerHour {
	134	+	ok, _, rel := l.Reserve(bg, "1.2.3.4")
	135	+	require.True(t, ok)
	136	+	releases = append(releases, rel)
	137	+	}
	138	+	ok, _, rel := l.Reserve(bg, "1.2.3.4")
	139	+	require.False(t, ok)
	140	+	rel() // no-op on a refusal
	141	+	releases[0]()
	142	+	releases[0]() // idempotent
	143	+	perAddr, global := l.Recent(bg, "1.2.3.4")
	144	+	assert.Equal(t, clientlimit.PerHour-1, perAddr)
	145	+	assert.Equal(t, clientlimit.PerHour-1, global)
	146	+	ok, _, _ = l.Reserve(bg, "1.2.3.4")
	147	+	assert.True(t, ok, "the refunded slot is usable again")
	148	+	}

 import (
 	"fmt"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 	t.Parallel()
 	l, clk := newSubs()
 	for i := range clientlimit.PerHour {
-		ok, _ := l.Allow(bg, "1.2.3.4")
+		ok, _, _ := l.Reserve(bg, "1.2.3.4")
 		require.True(t, ok, "submission %d", i)
-		l.Record(bg, "1.2.3.4")
 		clk.t = clk.t.Add(time.Minute)
 	}
-	ok, wait := l.Allow(bg, "1.2.3.4")
+	ok, wait, _ := l.Reserve(bg, "1.2.3.4")
 	assert.False(t, ok)
 	assert.Equal(t, time.Hour-3*time.Minute, wait, "until the first one leaves the hour")
-	other, _ := l.Allow(bg, "5.6.7.8")
+	other, _, _ := l.Reserve(bg, "5.6.7.8")
 	assert.True(t, other, "another client is not affected")
 	clk.t = clk.t.Add(wait)
-	ok, _ = l.Allow(bg, "1.2.3.4")
+	ok, _, _ = l.Reserve(bg, "1.2.3.4")
 	assert.True(t, ok, "the window slides")
 }
 	start := clk.t
 	// 3 per hour, spaced so the hourly limit never bites: one every 2 hours.
 	for range clientlimit.PerDay {
-		ok, _ := l.Allow(bg, "1.2.3.4")
+		ok, _, _ := l.Reserve(bg, "1.2.3.4")
 		require.True(t, ok)
-		l.Record(bg, "1.2.3.4")
 		clk.t = clk.t.Add(2 * time.Hour)
 	}
 	// 20h after start: hourly window empty, day window full.
-	ok, wait := l.Allow(bg, "1.2.3.4")
+	ok, wait, _ := l.Reserve(bg, "1.2.3.4")
 	assert.False(t, ok)
 	assert.Equal(t, start.Add(clientlimit.DayWindow).Sub(clk.t), wait)
 	clk.t = start.Add(clientlimit.DayWindow)
-	ok, _ = l.Allow(bg, "1.2.3.4")
+	ok, _, _ = l.Reserve(bg, "1.2.3.4")
 	assert.True(t, ok)
 }
 	t.Parallel()
 	l, _ := newSubs()
 	for i := range clientlimit.PerHour {
-		l.Record(bg, fmt.Sprintf("2001:db8:1:2::%x", i+1)) // same /64
+		l.Reserve(bg, fmt.Sprintf("2001:db8:1:2::%x", i+1)) // same /64
 	}
-	ok, _ := l.Allow(bg, "2001:db8:1:2:ffff::1")
+	ok, _, _ := l.Reserve(bg, "2001:db8:1:2:ffff::1")
 	assert.False(t, ok, "the whole /64 shares the budget")
-	ok, _ = l.Allow(bg, "2001:db8:1:3::1")
+	ok, _, _ = l.Reserve(bg, "2001:db8:1:3::1")
 	assert.True(t, ok, "another /64 does not")
-	ok, _ = l.Allow(bg, "::ffff:1.2.3.4")
+	ok, _, _ = l.Reserve(bg, "::ffff:1.2.3.4")
 	assert.True(t, ok)
-	l.Record(bg, "::ffff:1.2.3.4")
 	perAddr, _ := l.Recent(bg, "1.2.3.4")
 	assert.Equal(t, 1, perAddr, "a mapped address counts as its IPv4")
 }
 func TestSubmissionRecent(t *testing.T) {
 	t.Parallel()
 	l, clk := newSubs()
-	l.Record(bg, "1.1.1.1")
+	l.Reserve(bg, "1.1.1.1")
-	l.Record(bg, "1.1.1.1")
+	l.Reserve(bg, "1.1.1.1")
-	l.Record(bg, "2.2.2.2")
+	l.Reserve(bg, "2.2.2.2")
 	perAddr, global := l.Recent(bg, "1.1.1.1")
 	assert.Equal(t, 2, perAddr)
 	assert.Equal(t, 3, global)
 	l, clk := newSubs()
 	for i := range clientlimit.MaxTracked + 10 {
 		clk.t = clk.t.Add(time.Millisecond)
-		l.Record(bg, fmt.Sprintf("10.%d.%d.%d", i>>16&255, i>>8&255, i&255))
+		l.Reserve(bg, fmt.Sprintf("10.%d.%d.%d", i>>16&255, i>>8&255, i&255))
 	}
 	assert.LessOrEqual(t, l.Tracked(), clientlimit.MaxTracked, "bounded")
-	ok, _ := l.Allow(bg, "10.0.0.0")
+	ok, _, _ := l.Reserve(bg, "10.0.0.0")
 	assert.True(t, ok, "the oldest were evicted")
 }
+func TestSubmissionReserveIsAtomic(t *testing.T) {
+	t.Parallel()
+	l, _ := newSubs()
+	const n = 50
+	var granted atomic.Int32
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for range n {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			if ok, _, _ := l.Reserve(bg, "1.2.3.4"); ok {
+				granted.Add(1)
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+	assert.Equal(t, int32(clientlimit.PerHour), granted.Load())
+}
+func TestSubmissionReleaseRefunds(t *testing.T) {
+	t.Parallel()
+	l, _ := newSubs()
+	var releases []func()
+	for range clientlimit.PerHour {
+		ok, _, rel := l.Reserve(bg, "1.2.3.4")
+		require.True(t, ok)
+		releases = append(releases, rel)
+	}
+	ok, _, rel := l.Reserve(bg, "1.2.3.4")
+	require.False(t, ok)
+	rel() // no-op on a refusal
+	releases[0]()
+	releases[0]() // idempotent
+	perAddr, global := l.Recent(bg, "1.2.3.4")
+	assert.Equal(t, clientlimit.PerHour-1, perAddr)
+	assert.Equal(t, clientlimit.PerHour-1, global)
+	ok, _, _ = l.Reserve(bg, "1.2.3.4")
+	assert.True(t, ok, "the refunded slot is usable again")
+}

M internal/ports/mocks/SubmissionLimiter.go +42 -80


@@ -47,72 +47,6 @@ func (_m SubmissionLimiter) EXPECT() SubmissionLimiter_Expecter {
47	47		return &SubmissionLimiter_Expecter{mock: &_m.Mock}
48	48		}
49	49
50		-	// Allow provides a mock function for the type SubmissionLimiter
51		-	func (_mock *SubmissionLimiter) Allow(ctx context.Context, addr string) (bool, time.Duration) {
52		-	ret := _mock.Called(ctx, addr)
53		-
54		-	if len(ret) == 0 {
55		-	panic("no return value specified for Allow")
56		-	}
57		-
58		-	var r0 bool
59		-	var r1 time.Duration
60		-	if returnFunc, ok := ret.Get(0).(func(context.Context, string) (bool, time.Duration)); ok {
61		-	return returnFunc(ctx, addr)
62		-	}
63		-	if returnFunc, ok := ret.Get(0).(func(context.Context, string) bool); ok {
64		-	r0 = returnFunc(ctx, addr)
65		-	} else {
66		-	r0 = ret.Get(0).(bool)
67		-	}
68		-	if returnFunc, ok := ret.Get(1).(func(context.Context, string) time.Duration); ok {
69		-	r1 = returnFunc(ctx, addr)
70		-	} else {
71		-	r1 = ret.Get(1).(time.Duration)
72		-	}
73		-	return r0, r1
74		-	}
75		-
76		-	// SubmissionLimiter_Allow_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Allow'
77		-	type SubmissionLimiter_Allow_Call struct {
78		-	*mock.Call
79		-	}
80		-
81		-	// Allow is a helper method to define mock.On call
82		-	// - ctx context.Context
83		-	// - addr string
84		-	func (_e SubmissionLimiter_Expecter) Allow(ctx any, addr any) SubmissionLimiter_Allow_Call {
85		-	return &SubmissionLimiter_Allow_Call{Call: _e.mock.On("Allow", ctx, addr)}
86		-	}
87		-
88		-	func (_c SubmissionLimiter_Allow_Call) Run(run func(ctx context.Context, addr string)) SubmissionLimiter_Allow_Call {
89		-	_c.Call.Run(func(args mock.Arguments) {
90		-	var arg0 context.Context
91		-	if args[0] != nil {
92		-	arg0 = args[0].(context.Context)
93		-	}
94		-	var arg1 string
95		-	if args[1] != nil {
96		-	arg1 = args[1].(string)
97		-	}
98		-	run(
99		-	arg0,
100		-	arg1,
101		-	)
102		-	})
103		-	return _c
104		-	}
105		-
106		-	func (_c SubmissionLimiter_Allow_Call) Return(ok bool, retryAfter time.Duration) SubmissionLimiter_Allow_Call {
107		-	_c.Call.Return(ok, retryAfter)
108		-	return _c
109		-	}
110		-
111		-	func (_c SubmissionLimiter_Allow_Call) RunAndReturn(run func(ctx context.Context, addr string) (bool, time.Duration)) SubmissionLimiter_Allow_Call {
112		-	_c.Call.Return(run)
113		-	return _c
114		-	}
115		-
116	50		// Recent provides a mock function for the type SubmissionLimiter
117	51		func (_mock *SubmissionLimiter) Recent(ctx context.Context, addr string) (int, int) {
118	52		ret := _mock.Called(ctx, addr)

@@ -179,25 +113,53 @@ func (_c *SubmissionLimiter_Recent_Call) RunAndReturn(run func(ctx context.Conte
179	113		return _c
180	114		}
181	115
182		-	// Record provides a mock function for the type SubmissionLimiter
183		-	func (_mock *SubmissionLimiter) Record(ctx context.Context, addr string) {
184		-	_mock.Called(ctx, addr)
185		-	return
	116	+	// Reserve provides a mock function for the type SubmissionLimiter
	117	+	func (_mock *SubmissionLimiter) Reserve(ctx context.Context, addr string) (bool, time.Duration, func()) {
	118	+	ret := _mock.Called(ctx, addr)
	119	+
	120	+	if len(ret) == 0 {
	121	+	panic("no return value specified for Reserve")
	122	+	}
	123	+
	124	+	var r0 bool
	125	+	var r1 time.Duration
	126	+	var r2 func()
	127	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string) (bool, time.Duration, func())); ok {
	128	+	return returnFunc(ctx, addr)
	129	+	}
	130	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string) bool); ok {
	131	+	r0 = returnFunc(ctx, addr)
	132	+	} else {
	133	+	r0 = ret.Get(0).(bool)
	134	+	}
	135	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string) time.Duration); ok {
	136	+	r1 = returnFunc(ctx, addr)
	137	+	} else {
	138	+	r1 = ret.Get(1).(time.Duration)
	139	+	}
	140	+	if returnFunc, ok := ret.Get(2).(func(context.Context, string) func()); ok {
	141	+	r2 = returnFunc(ctx, addr)
	142	+	} else {
	143	+	if ret.Get(2) != nil {
	144	+	r2 = ret.Get(2).(func())
	145	+	}
	146	+	}
	147	+	return r0, r1, r2
186	148		}
187	149
188		-	// SubmissionLimiter_Record_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Record'
189		-	type SubmissionLimiter_Record_Call struct {
	150	+	// SubmissionLimiter_Reserve_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Reserve'
	151	+	type SubmissionLimiter_Reserve_Call struct {
190	152		*mock.Call
191	153		}
192	154
193		-	// Record is a helper method to define mock.On call
	155	+	// Reserve is a helper method to define mock.On call
194	156		// - ctx context.Context
195	157		// - addr string
196		-	func (_e SubmissionLimiter_Expecter) Record(ctx any, addr any) SubmissionLimiter_Record_Call {
197		-	return &SubmissionLimiter_Record_Call{Call: _e.mock.On("Record", ctx, addr)}
	158	+	func (_e SubmissionLimiter_Expecter) Reserve(ctx any, addr any) SubmissionLimiter_Reserve_Call {
	159	+	return &SubmissionLimiter_Reserve_Call{Call: _e.mock.On("Reserve", ctx, addr)}
198	160		}
199	161
200		-	func (_c SubmissionLimiter_Record_Call) Run(run func(ctx context.Context, addr string)) SubmissionLimiter_Record_Call {
	162	+	func (_c SubmissionLimiter_Reserve_Call) Run(run func(ctx context.Context, addr string)) SubmissionLimiter_Reserve_Call {
201	163		_c.Call.Run(func(args mock.Arguments) {
202	164		var arg0 context.Context
203	165		if args[0] != nil {

@@ -215,12 +177,12 @@ func (_c *SubmissionLimiter_Record_Call) Run(run func(ctx context.Context, addr
215	177		return _c
216	178		}
217	179
218		-	func (_c SubmissionLimiter_Record_Call) Return() SubmissionLimiter_Record_Call {
219		-	_c.Call.Return()
	180	+	func (_c SubmissionLimiter_Reserve_Call) Return(ok bool, retryAfter time.Duration, release func()) SubmissionLimiter_Reserve_Call {
	181	+	_c.Call.Return(ok, retryAfter, release)
220	182		return _c
221	183		}
222	184
223		-	func (_c SubmissionLimiter_Record_Call) RunAndReturn(run func(ctx context.Context, addr string)) SubmissionLimiter_Record_Call {
224		-	_c.Run(run)
	185	+	func (_c SubmissionLimiter_Reserve_Call) RunAndReturn(run func(ctx context.Context, addr string) (bool, time.Duration, func())) SubmissionLimiter_Reserve_Call {
	186	+	_c.Call.Return(run)
225	187		return _c
226	188		}


@@ -47,72 +47,6 @@ func (_m SubmissionLimiter) EXPECT() SubmissionLimiter_Expecter {
47		return &SubmissionLimiter_Expecter{mock: &_m.Mock}	47		return &SubmissionLimiter_Expecter{mock: &_m.Mock}
48		}	48		}
49			49
50	-	// Allow provides a mock function for the type SubmissionLimiter
51	-	func (_mock *SubmissionLimiter) Allow(ctx context.Context, addr string) (bool, time.Duration) {
52	-	ret := _mock.Called(ctx, addr)
53	-
54	-	if len(ret) == 0 {
55	-	panic("no return value specified for Allow")
56	-	}
57	-
58	-	var r0 bool
59	-	var r1 time.Duration
60	-	if returnFunc, ok := ret.Get(0).(func(context.Context, string) (bool, time.Duration)); ok {
61	-	return returnFunc(ctx, addr)
62	-	}
63	-	if returnFunc, ok := ret.Get(0).(func(context.Context, string) bool); ok {
64	-	r0 = returnFunc(ctx, addr)
65	-	} else {
66	-	r0 = ret.Get(0).(bool)
67	-	}
68	-	if returnFunc, ok := ret.Get(1).(func(context.Context, string) time.Duration); ok {
69	-	r1 = returnFunc(ctx, addr)
70	-	} else {
71	-	r1 = ret.Get(1).(time.Duration)
72	-	}
73	-	return r0, r1
74	-	}
75	-
76	-	// SubmissionLimiter_Allow_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Allow'
77	-	type SubmissionLimiter_Allow_Call struct {
78	-	*mock.Call
79	-	}
80	-
81	-	// Allow is a helper method to define mock.On call
82	-	// - ctx context.Context
83	-	// - addr string
84	-	func (_e SubmissionLimiter_Expecter) Allow(ctx any, addr any) SubmissionLimiter_Allow_Call {
85	-	return &SubmissionLimiter_Allow_Call{Call: _e.mock.On("Allow", ctx, addr)}
86	-	}
87	-
88	-	func (_c SubmissionLimiter_Allow_Call) Run(run func(ctx context.Context, addr string)) SubmissionLimiter_Allow_Call {
89	-	_c.Call.Run(func(args mock.Arguments) {
90	-	var arg0 context.Context
91	-	if args[0] != nil {
92	-	arg0 = args[0].(context.Context)
93	-	}
94	-	var arg1 string
95	-	if args[1] != nil {
96	-	arg1 = args[1].(string)
97	-	}
98	-	run(
99	-	arg0,
100	-	arg1,
101	-	)
102	-	})
103	-	return _c
104	-	}
105	-
106	-	func (_c SubmissionLimiter_Allow_Call) Return(ok bool, retryAfter time.Duration) SubmissionLimiter_Allow_Call {
107	-	_c.Call.Return(ok, retryAfter)
108	-	return _c
109	-	}
110	-
111	-	func (_c SubmissionLimiter_Allow_Call) RunAndReturn(run func(ctx context.Context, addr string) (bool, time.Duration)) SubmissionLimiter_Allow_Call {
112	-	_c.Call.Return(run)
113	-	return _c
114	-	}
115	-
116		// Recent provides a mock function for the type SubmissionLimiter	50		// Recent provides a mock function for the type SubmissionLimiter
117		func (_mock *SubmissionLimiter) Recent(ctx context.Context, addr string) (int, int) {	51		func (_mock *SubmissionLimiter) Recent(ctx context.Context, addr string) (int, int) {
118		ret := _mock.Called(ctx, addr)	52		ret := _mock.Called(ctx, addr)

@@ -179,25 +113,53 @@ func (_c *SubmissionLimiter_Recent_Call) RunAndReturn(run func(ctx context.Conte
179		return _c	113		return _c
180		}	114		}
181			115
182	-	// Record provides a mock function for the type SubmissionLimiter	116	+	// Reserve provides a mock function for the type SubmissionLimiter
183	-	func (_mock *SubmissionLimiter) Record(ctx context.Context, addr string) {	117	+	func (_mock *SubmissionLimiter) Reserve(ctx context.Context, addr string) (bool, time.Duration, func()) {
184	-	_mock.Called(ctx, addr)	118	+	ret := _mock.Called(ctx, addr)
185	-	return	119	+
			120	+	if len(ret) == 0 {
			121	+	panic("no return value specified for Reserve")
			122	+	}
			123	+
			124	+	var r0 bool
			125	+	var r1 time.Duration
			126	+	var r2 func()
			127	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string) (bool, time.Duration, func())); ok {
			128	+	return returnFunc(ctx, addr)
			129	+	}
			130	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string) bool); ok {
			131	+	r0 = returnFunc(ctx, addr)
			132	+	} else {
			133	+	r0 = ret.Get(0).(bool)
			134	+	}
			135	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string) time.Duration); ok {
			136	+	r1 = returnFunc(ctx, addr)
			137	+	} else {
			138	+	r1 = ret.Get(1).(time.Duration)
			139	+	}
			140	+	if returnFunc, ok := ret.Get(2).(func(context.Context, string) func()); ok {
			141	+	r2 = returnFunc(ctx, addr)
			142	+	} else {
			143	+	if ret.Get(2) != nil {
			144	+	r2 = ret.Get(2).(func())
			145	+	}
			146	+	}
			147	+	return r0, r1, r2
186		}	148		}
187			149
188	-	// SubmissionLimiter_Record_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Record'	150	+	// SubmissionLimiter_Reserve_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Reserve'
189	-	type SubmissionLimiter_Record_Call struct {	151	+	type SubmissionLimiter_Reserve_Call struct {
190		*mock.Call	152		*mock.Call
191		}	153		}
192			154
193	-	// Record is a helper method to define mock.On call	155	+	// Reserve is a helper method to define mock.On call
194		// - ctx context.Context	156		// - ctx context.Context
195		// - addr string	157		// - addr string
196	-	func (_e SubmissionLimiter_Expecter) Record(ctx any, addr any) SubmissionLimiter_Record_Call {	158	+	func (_e SubmissionLimiter_Expecter) Reserve(ctx any, addr any) SubmissionLimiter_Reserve_Call {
197	-	return &SubmissionLimiter_Record_Call{Call: _e.mock.On("Record", ctx, addr)}	159	+	return &SubmissionLimiter_Reserve_Call{Call: _e.mock.On("Reserve", ctx, addr)}
198		}	160		}
199			161
200	-	func (_c SubmissionLimiter_Record_Call) Run(run func(ctx context.Context, addr string)) SubmissionLimiter_Record_Call {	162	+	func (_c SubmissionLimiter_Reserve_Call) Run(run func(ctx context.Context, addr string)) SubmissionLimiter_Reserve_Call {
201		_c.Call.Run(func(args mock.Arguments) {	163		_c.Call.Run(func(args mock.Arguments) {
202		var arg0 context.Context	164		var arg0 context.Context
203		if args[0] != nil {	165		if args[0] != nil {

@@ -215,12 +177,12 @@ func (_c *SubmissionLimiter_Record_Call) Run(run func(ctx context.Context, addr
215		return _c	177		return _c
216		}	178		}
217			179
218	-	func (_c SubmissionLimiter_Record_Call) Return() SubmissionLimiter_Record_Call {	180	+	func (_c SubmissionLimiter_Reserve_Call) Return(ok bool, retryAfter time.Duration, release func()) SubmissionLimiter_Reserve_Call {
219	-	_c.Call.Return()	181	+	_c.Call.Return(ok, retryAfter, release)
220		return _c	182		return _c
221		}	183		}
222			184
223	-	func (_c SubmissionLimiter_Record_Call) RunAndReturn(run func(ctx context.Context, addr string)) SubmissionLimiter_Record_Call {	185	+	func (_c SubmissionLimiter_Reserve_Call) RunAndReturn(run func(ctx context.Context, addr string) (bool, time.Duration, func())) SubmissionLimiter_Reserve_Call {
224	-	_c.Run(run)	186	+	_c.Call.Return(run)
225		return _c	187		return _c
226		}	188		}

M internal/ports/services.go +5 -4


@@ -321,10 +321,11 @@ type FailureLimiter interface {
321	321		// SubmissionLimiter limits the submissions of the public issue form per
322	322		// client. *clientlimit.SubmissionLimiter implements it.
323	323		type SubmissionLimiter interface {
324		-	// Allow reports whether addr may submit now, and when to retry if not.
325		-	Allow(ctx context.Context, addr string) (ok bool, retryAfter time.Duration)
326		-	// Record counts a submission from addr.
327		-	Record(ctx context.Context, addr string)
	324	+	// Reserve atomically checks the limits for addr and, if it may submit,
	325	+	// counts a submission. When it may not, retryAfter says when to retry.
	326	+	// release refunds the reservation (only when the submission was not
	327	+	// stored); it is never nil and safe to call once.
	328	+	Reserve(ctx context.Context, addr string) (ok bool, retryAfter time.Duration, release func())
328	329		// Recent returns the submissions of the last hour from addr and from all
329	330		// clients.
330	331		Recent(ctx context.Context, addr string) (perAddr, global int)

 // SubmissionLimiter limits the submissions of the public issue form per
 // client. *clientlimit.SubmissionLimiter implements it.
 type SubmissionLimiter interface {
-	// Allow reports whether addr may submit now, and when to retry if not.
+	// Reserve atomically checks the limits for addr and, if it may submit,
-	Allow(ctx context.Context, addr string) (ok bool, retryAfter time.Duration)
+	// counts a submission. When it may not, retryAfter says when to retry.
-	// Record counts a submission from addr.
+	// release refunds the reservation (only when the submission was not
-	Record(ctx context.Context, addr string)
+	// stored); it is never nil and safe to call once.
+	Reserve(ctx context.Context, addr string) (ok bool, retryAfter time.Duration, release func())
 	// Recent returns the submissions of the last hour from addr and from all
 	// clients.
 	Recent(ctx context.Context, addr string) (perAddr, global int)

M internal/usecase/issue_submissions.go +39 -22


@@ -106,6 +106,8 @@ type IssueSubmissions struct {
106	106		limit int
107	107		minFill time.Duration
108	108
	109	+	admitMu sync.Mutex // serializes the pending cap check and the store
	110	+
109	111		mu sync.Mutex
110	112		spent map[string]time.Time // nonce -> expiry of its challenge
111	113		nextSweep time.Time

@@ -284,23 +286,6 @@ func (s *IssueSubmissions) Submit(ctx context.Context, repo domain.RepoName, add
284	286		}
285	287		ipHash := s.keys.IPHash(addr)
286	288		now := s.clock.Now().UTC()
287		-	banned, err := s.bans.Banned(ctx, ipHash, now)
288		-	if err != nil {
289		-	return fmt.Errorf("check ban: %w", err)
290		-	}
291		-	if banned {
292		-	return refuse("banned", domain.ErrSubmissionInvalid)
293		-	}
294		-	if ok, after := s.limiter.Allow(ctx, addr); !ok {
295		-	return refuse("rate_limited", &domain.SubmissionRetryError{After: after})
296		-	}
297		-	n, err := s.pending.Count(ctx)
298		-	if err != nil {
299		-	return fmt.Errorf("count pending issues: %w", err)
300		-	}
301		-	if n >= domain.MaxPendingIssues {
302		-	return refuse("pending_full", domain.ErrSubmissionsPaused)
303		-	}
304	289		if f.Honeypot != "" {
305	290		return refuse("honeypot", domain.ErrSubmissionInvalid)
306	291		}

@@ -335,6 +320,13 @@ func (s *IssueSubmissions) Submit(ctx context.Context, repo domain.RepoName, add
335	320		}
336	321		return refuse("replayed", err)
337	322		}
	323	+	banned, err := s.bans.Banned(ctx, ipHash, now)
	324	+	if err != nil {
	325	+	return fmt.Errorf("check ban: %w", err)
	326	+	}
	327	+	if banned {
	328	+	return refuse("banned", domain.ErrSubmissionInvalid)
	329	+	}
338	330		title, err := domain.ParseIssueTitle(f.Title)
339	331		if err != nil {
340	332		return refuse("text_title", fmt.Errorf("title: %w", err))

@@ -351,16 +343,41 @@ func (s *IssueSubmissions) Submit(ctx context.Context, repo domain.RepoName, add
351	343		if err != nil {
352	344		return fmt.Errorf("new pending issue id: %w", err)
353	345		}
354		-	s.limiter.Record(ctx, addr)
355		-	err = s.pending.Put(ctx, domain.PendingIssue{
	346	+	if err := s.admit(ctx, addr, refuse, domain.PendingIssue{
356	347		ID: id, Repo: repo, Title: title, Body: body, Author: author, CreatedAt: now, IPHash: ipHash,
357		-	})
358		-	if err != nil {
359		-	return fmt.Errorf("store pending issue of %q: %w", repo, err)
	348	+	}); err != nil {
	349	+	return err
360	350		}
361	351		log.Info().Str("repo", repo.String()).Msg("issue submitted for moderation")
362	352		s.bus.Publish(ctx, domain.IssueSubmitted{Repo: repo, ID: id, At: now})
363	353		return nil
364	354		}
365	355
	356	+	// admit reserves the client's quota, checks the global queue cap and stores
	357	+	// the issue. Ermite is a single process (ADR 0001), so admitMu serializes the
	358	+	// Count -> Put section: concurrent submissions cannot exceed
	359	+	// domain.MaxPendingIssues. It would not hold with several processes.
	360	+	func (s *IssueSubmissions) admit(ctx context.Context, addr string, refuse func(string, error) error, p domain.PendingIssue) error {
	361	+	s.admitMu.Lock()
	362	+	defer s.admitMu.Unlock()
	363	+	ok, after, release := s.limiter.Reserve(ctx, addr)
	364	+	if !ok {
	365	+	return refuse("rate_limited", &domain.SubmissionRetryError{After: after})
	366	+	}
	367	+	n, err := s.pending.Count(ctx)
	368	+	if err != nil {
	369	+	release()
	370	+	return fmt.Errorf("count pending issues: %w", err)
	371	+	}
	372	+	if n >= domain.MaxPendingIssues {
	373	+	release()
	374	+	return refuse("pending_full", domain.ErrSubmissionsPaused)
	375	+	}
	376	+	if err := s.pending.Put(ctx, p); err != nil {
	377	+	release()
	378	+	return fmt.Errorf("store pending issue of %q: %w", p.Repo, err)
	379	+	}
	380	+	return nil
	381	+	}
	382	+
366	383		func enc64(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) }

 	limit   int
 	minFill time.Duration
+	admitMu sync.Mutex // serializes the pending cap check and the store
 	mu        sync.Mutex
 	spent     map[string]time.Time // nonce -> expiry of its challenge
 	nextSweep time.Time
 	}
 	ipHash := s.keys.IPHash(addr)
 	now := s.clock.Now().UTC()
-	banned, err := s.bans.Banned(ctx, ipHash, now)
-	if err != nil {
-		return fmt.Errorf("check ban: %w", err)
-	}
-	if banned {
-		return refuse("banned", domain.ErrSubmissionInvalid)
-	}
-	if ok, after := s.limiter.Allow(ctx, addr); !ok {
-		return refuse("rate_limited", &domain.SubmissionRetryError{After: after})
-	}
-	n, err := s.pending.Count(ctx)
-	if err != nil {
-		return fmt.Errorf("count pending issues: %w", err)
-	}
-	if n >= domain.MaxPendingIssues {
-		return refuse("pending_full", domain.ErrSubmissionsPaused)
-	}
 	if f.Honeypot != "" {
 		return refuse("honeypot", domain.ErrSubmissionInvalid)
 	}
 		}
 		return refuse("replayed", err)
 	}
+	banned, err := s.bans.Banned(ctx, ipHash, now)
+	if err != nil {
+		return fmt.Errorf("check ban: %w", err)
+	}
+	if banned {
+		return refuse("banned", domain.ErrSubmissionInvalid)
+	}
 	title, err := domain.ParseIssueTitle(f.Title)
 	if err != nil {
 		return refuse("text_title", fmt.Errorf("title: %w", err))
 	if err != nil {
 		return fmt.Errorf("new pending issue id: %w", err)
 	}
-	s.limiter.Record(ctx, addr)
+	if err := s.admit(ctx, addr, refuse, domain.PendingIssue{
-	err = s.pending.Put(ctx, domain.PendingIssue{
 		ID: id, Repo: repo, Title: title, Body: body, Author: author, CreatedAt: now, IPHash: ipHash,
-	})
+	}); err != nil {
-	if err != nil {
+		return err
-		return fmt.Errorf("store pending issue of %q: %w", repo, err)
 	}
 	log.Info().Str("repo", repo.String()).Msg("issue submitted for moderation")
 	s.bus.Publish(ctx, domain.IssueSubmitted{Repo: repo, ID: id, At: now})
 	return nil
 }
+// admit reserves the client's quota, checks the global queue cap and stores
+// the issue. Ermite is a single process (ADR 0001), so admitMu serializes the
+// Count -> Put section: concurrent submissions cannot exceed
+// domain.MaxPendingIssues. It would not hold with several processes.
+func (s *IssueSubmissions) admit(ctx context.Context, addr string, refuse func(string, error) error, p domain.PendingIssue) error {
+	s.admitMu.Lock()
+	defer s.admitMu.Unlock()
+	ok, after, release := s.limiter.Reserve(ctx, addr)
+	if !ok {
+		return refuse("rate_limited", &domain.SubmissionRetryError{After: after})
+	}
+	n, err := s.pending.Count(ctx)
+	if err != nil {
+		release()
+		return fmt.Errorf("count pending issues: %w", err)
+	}
+	if n >= domain.MaxPendingIssues {
+		release()
+		return refuse("pending_full", domain.ErrSubmissionsPaused)
+	}
+	if err := s.pending.Put(ctx, p); err != nil {
+		release()
+		return fmt.Errorf("store pending issue of %q: %w", p.Repo, err)
+	}
+	return nil
+}
 func enc64(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) }

M internal/usecase/issue_submissions_test.go +208 -9


@@ -11,6 +11,7 @@ import (
11	11		"strconv"
12	12		"strings"
13	13		"sync"
	14	+	"sync/atomic"
14	15		"testing"
15	16		"time"
16	17

@@ -73,11 +74,16 @@ func newSubFixture(t testing.T) subFixture {
73	74		return f
74	75		}
75	76
76		-	// open expects the checks before the token: not banned, allowed, room.
77		-	func (f *subFixture) open(pendingCount int) {
	77	+	// open expects the checks after the proof of work: not banned, quota
	78	+	// available, room. It returns the number of reservations released.
	79	+	func (f subFixture) open(pendingCount int) atomic.Int32 {
	80	+	released := new(atomic.Int32)
78	81		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()
79		-	f.limiter.EXPECT().Allow(mock.Anything, mock.Anything).Return(true, time.Duration(0)).Maybe()
	82	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, string) (bool, time.Duration, func()) {
	83	+	return true, 0, func() { released.Add(1) }
	84	+	}).Maybe()
80	85		f.pending.EXPECT().Count(mock.Anything).Return(pendingCount, nil).Maybe()
	86	+	return released
81	87		}
82	88
83	89		func (f *subFixture) challenge(addr string) ports.IssueChallenge {

@@ -179,7 +185,6 @@ func TestIssueSubmitSuccess(t *testing.T) {
179	185		ch := f.challenge(f.addr)
180	186		sub := f.valid(ch)
181	187		f.advance(usecase.MinFillTime)
182		-	f.limiter.EXPECT().Record(mock.Anything, f.addr).Once()
183	188		var put domain.PendingIssue
184	189		f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, p domain.PendingIssue) error {
185	190		put = p

@@ -217,11 +222,11 @@ func TestIssueSubmitRefusals(t *testing.T) {
217	222		}},
218	223		{name: "rate limited", want: domain.ErrSubmissionRateLimited, wait: wait, setup: func(f *subFixture) {
219	224		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
220		-	f.limiter.EXPECT().Allow(mock.Anything, mock.Anything).Return(false, 5*time.Minute)
	225	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(false, 5*time.Minute, func() {})
221	226		}},
222	227		{name: "paused at 200", want: domain.ErrSubmissionsPaused, wait: wait, setup: func(f *subFixture) {
223	228		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
224		-	f.limiter.EXPECT().Allow(mock.Anything, mock.Anything).Return(true, time.Duration(0))
	229	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(true, time.Duration(0), func() {})
225	230		f.pending.EXPECT().Count(mock.Anything).Return(domain.MaxPendingIssues, nil)
226	231		}},
227	232		{name: "honeypot", want: domain.ErrSubmissionInvalid, wait: wait, arrange: func(_ subFixture, _ ports.IssueChallenge, s *ports.IssueSubmission) string {

@@ -326,7 +331,7 @@ func TestIssueSubmitReplayedNonce(t *testing.T) {
326	331		ch := f.challenge(f.addr)
327	332		sub := f.valid(ch)
328	333		f.advance(usecase.MinFillTime)
329		-	// first: text refused, nonce is spent anyway
	334	+	// first: text refused, nonce is spent anyway (nothing reserved)
330	335		bad := sub
331	336		bad.Title = ""
332	337		var te *domain.TextError

@@ -344,7 +349,6 @@ func TestIssueSubmitReplayAfterSuccess(t *testing.T) {
344	349		ch := f.challenge(f.addr)
345	350		sub := f.valid(ch)
346	351		f.advance(usecase.MinFillTime)
347		-	f.limiter.EXPECT().Record(mock.Anything, f.addr).Once()
348	352		f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Once()
349	353		f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Once()
350	354		require.NoError(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub))

@@ -390,7 +394,6 @@ func TestIssueSubmitSpentSetExpiry(t *testing.T) {
390	394		t.Parallel()
391	395		f := newSubFixture(t)
392	396		f.open(0)
393		-	f.limiter.EXPECT().Record(mock.Anything, mock.Anything).Maybe()
394	397		f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Maybe()
395	398		f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()
396	399		ch := f.challenge(f.addr)

@@ -469,3 +472,199 @@ func TestDifficulty(t *testing.T) {
469	472		assert.Equal(t, 18, usecase.BaseDifficulty)
470	473		assert.Equal(t, 21, usecase.MaxDifficulty)
471	474		}
	475	+
	476	+	// submitters builds n valid, distinct submissions (one challenge each) for the
	477	+	// addresses addrOf(i) and the repositories repoOf(i), ready to be sent.
	478	+	type prepared struct {
	479	+	repo domain.RepoName
	480	+	addr string
	481	+	sub ports.IssueSubmission
	482	+	}
	483	+
	484	+	func (f *subFixture) prepare(n int, addrOf func(int) string, repoOf func(int) domain.RepoName) []prepared {
	485	+	out := make([]prepared, n)
	486	+	for i := range n {
	487	+	addr := addrOf(i)
	488	+	repo := repoOf(i)
	489	+	f.limiter.EXPECT().Recent(mock.Anything, addr).Return(0, 0).Once()
	490	+	ch, err := f.uc.Challenge(context.Background(), repo, addr)
	491	+	require.NoError(f.t, err)
	492	+	out[i] = prepared{repo: repo, addr: addr, sub: f.valid(ch)}
	493	+	}
	494	+	f.advance(usecase.MinFillTime)
	495	+	return out
	496	+	}
	497	+
	498	+	// race sends every prepared submission at once and returns the errors.
	499	+	func (f *subFixture) race(ps []prepared) []error {
	500	+	errs := make([]error, len(ps))
	501	+	var wg sync.WaitGroup
	502	+	start := make(chan struct{})
	503	+	for i, p := range ps {
	504	+	wg.Add(1)
	505	+	go func() {
	506	+	defer wg.Done()
	507	+	<-start
	508	+	errs[i] = f.uc.Submit(context.Background(), p.repo, p.addr, p.sub)
	509	+	}()
	510	+	}
	511	+	close(start)
	512	+	wg.Wait()
	513	+	return errs
	514	+	}
	515	+
	516	+	func TestIssueSubmitConcurrentSameAddrRespectsQuota(t *testing.T) {
	517	+	t.Parallel()
	518	+	f := newSubFixture(t)
	519	+	f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()
	520	+	f.pending.EXPECT().Count(mock.Anything).Return(0, nil).Maybe()
	521	+	// a real, atomic limiter semantics: PerHour grants then refusals
	522	+	var granted atomic.Int32
	523	+	var rlMu sync.Mutex
	524	+	f.limiter.EXPECT().Reserve(mock.Anything, f.addr).RunAndReturn(func(context.Context, string) (bool, time.Duration, func()) {
	525	+	rlMu.Lock()
	526	+	defer rlMu.Unlock()
	527	+	if granted.Load() >= 3 {
	528	+	return false, time.Minute, func() {}
	529	+	}
	530	+	granted.Add(1)
	531	+	return true, 0, func() {}
	532	+	}).Maybe()
	533	+	var puts atomic.Int32
	534	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, domain.PendingIssue) error {
	535	+	puts.Add(1)
	536	+	return nil
	537	+	}).Maybe()
	538	+	f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()
	539	+	ps := f.prepare(20, func(int) string { return f.addr }, func(int) domain.RepoName { return f.repo.Name })
	540	+	var ok, limited int
	541	+	for _, err := range f.race(ps) {
	542	+	var re *domain.SubmissionRetryError
	543	+	switch {
	544	+	case err == nil:
	545	+	ok++
	546	+	case errors.As(err, &re):
	547	+	limited++
	548	+	default:
	549	+	t.Fatalf("unexpected error: %v", err)
	550	+	}
	551	+	}
	552	+	assert.Equal(t, 3, ok)
	553	+	assert.Equal(t, 17, limited)
	554	+	assert.Equal(t, int32(3), puts.Load())
	555	+	}
	556	+
	557	+	func TestIssueSubmitConcurrentRespectsPendingCap(t *testing.T) {
	558	+	t.Parallel()
	559	+	f := newSubFixture(t)
	560	+	f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()
	561	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(true, time.Duration(0), func() {}).Maybe()
	562	+	other := name(t, "other")
	563	+	f.meta.EXPECT().Get(mock.Anything, other).Return(domain.Repo{Name: other, Visibility: domain.VisibilityPublic, Settings: withIssues(domain.IssuesPublic)}, nil).Maybe()
	564	+	var count atomic.Int32
	565	+	count.Store(domain.MaxPendingIssues - 1)
	566	+	f.pending.EXPECT().Count(mock.Anything).RunAndReturn(func(context.Context) (int, error) {
	567	+	return int(count.Load()), nil
	568	+	}).Maybe()
	569	+	var puts atomic.Int32
	570	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, domain.PendingIssue) error {
	571	+	puts.Add(1)
	572	+	count.Add(1)
	573	+	return nil
	574	+	}).Maybe()
	575	+	f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()
	576	+	ps := f.prepare(20, func(i int) string { return "ip:198.51.100." + strconv.Itoa(i+1) }, func(i int) domain.RepoName {
	577	+	if i%2 == 0 {
	578	+	return f.repo.Name
	579	+	}
	580	+	return other
	581	+	})
	582	+	var ok, paused int
	583	+	for _, err := range f.race(ps) {
	584	+	switch {
	585	+	case err == nil:
	586	+	ok++
	587	+	case errors.Is(err, domain.ErrSubmissionsPaused):
	588	+	paused++
	589	+	default:
	590	+	t.Fatalf("unexpected error: %v", err)
	591	+	}
	592	+	}
	593	+	assert.Equal(t, 1, ok)
	594	+	assert.Equal(t, 19, paused)
	595	+	assert.Equal(t, int32(1), puts.Load())
	596	+	}
	597	+
	598	+	func TestIssueSubmitReleasesTheReservationWhenStoreFails(t *testing.T) {
	599	+	t.Parallel()
	600	+	f := newSubFixture(t)
	601	+	released := f.open(0)
	602	+	boom := errors.New("disk full")
	603	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(boom).Once()
	604	+	ch := f.challenge(f.addr)
	605	+	sub := f.valid(ch)
	606	+	f.advance(usecase.MinFillTime)
	607	+	require.ErrorIs(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub), boom)
	608	+	assert.Equal(t, int32(1), released.Load())
	609	+
	610	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Once()
	611	+	f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Once()
	612	+	ch = f.challenge(f.addr)
	613	+	sub = f.valid(ch)
	614	+	f.advance(usecase.MinFillTime)
	615	+	require.NoError(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub))
	616	+	assert.Equal(t, int32(1), released.Load(), "a stored submission keeps its reservation")
	617	+	}
	618	+
	619	+	func TestIssueSubmitReleasesTheReservationWhenPaused(t *testing.T) {
	620	+	t.Parallel()
	621	+	f := newSubFixture(t)
	622	+	released := f.open(domain.MaxPendingIssues)
	623	+	ch := f.challenge(f.addr)
	624	+	sub := f.valid(ch)
	625	+	f.advance(usecase.MinFillTime)
	626	+	assert.ErrorIs(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub), domain.ErrSubmissionsPaused)
	627	+	assert.Equal(t, int32(1), released.Load())
	628	+	}
	629	+
	630	+	// TestIssueSubmitNoStateOracleBeforeProofOfWork: a banned client, a limited
	631	+	// one, or a full queue all get the same answer to an invalid submission, and
	632	+	// no limiter, count or store is touched (the mocks have no expectation).
	633	+	func TestIssueSubmitNoStateOracleBeforeProofOfWork(t *testing.T) {
	634	+	t.Parallel()
	635	+	f := newSubFixture(t)
	636	+	ch := f.challenge(f.addr)
	637	+	valid := f.valid(ch)
	638	+	f.advance(usecase.MinFillTime)
	639	+	bad := []ports.IssueSubmission{
	640	+	{},
	641	+	{Token: "garbage", Solution: "x", FormToken: "y", Title: "", Body: ""},
	642	+	{Token: ch.Token, Honeypot: "x"},
	643	+	{Token: ch.Token, FormToken: ch.FormToken, Solution: "nope"},
	644	+	{Token: valid.Token, FormToken: valid.FormToken, Solution: "abc", Title: "two\nlines"}, // text error + bad pow
	645	+	}
	646	+	for _, s := range bad {
	647	+	err := f.uc.Submit(context.Background(), f.repo.Name, f.addr, s)
	648	+	require.ErrorIs(t, err, domain.ErrSubmissionInvalid)
	649	+	assert.NotErrorIs(t, err, domain.ErrInvalidText)
	650	+	}
	651	+	// a fully valid submission from a banned client: still invalid, not
	652	+	// limited, paused or a text error, even with bad text.
	653	+	f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(true, nil).Once()
	654	+	bannedBadText := valid
	655	+	bannedBadText.Title = ""
	656	+	err := f.uc.Submit(context.Background(), f.repo.Name, f.addr, bannedBadText)
	657	+	require.ErrorIs(t, err, domain.ErrSubmissionInvalid)
	658	+	assert.NotErrorIs(t, err, domain.ErrInvalidText)
	659	+	var re *domain.SubmissionRetryError
	660	+	assert.False(t, errors.As(err, &re))
	661	+	assert.NotErrorIs(t, err, domain.ErrSubmissionsPaused)
	662	+	}
	663	+
	664	+	func TestIssueChallengeRevealsNoState(t *testing.T) {
	665	+	t.Parallel()
	666	+	f := newSubFixture(t) // no Banned, Reserve or Count expectation
	667	+	f.limiter.EXPECT().Recent(mock.Anything, f.addr).Return(50, 500).Once()
	668	+	_, err := f.uc.Challenge(context.Background(), f.repo.Name, f.addr)
	669	+	require.NoError(t, err)
	670	+	}


@@ -11,6 +11,7 @@ import (
11		"strconv"	11		"strconv"
12		"strings"	12		"strings"
13		"sync"	13		"sync"
			14	+	"sync/atomic"
14		"testing"	15		"testing"
15		"time"	16		"time"
16			17

@@ -73,11 +74,16 @@ func newSubFixture(t testing.T) subFixture {
73		return f	74		return f
74		}	75		}
75			76
76	-	// open expects the checks before the token: not banned, allowed, room.	77	+	// open expects the checks after the proof of work: not banned, quota
77	-	func (f *subFixture) open(pendingCount int) {	78	+	// available, room. It returns the number of reservations released.
			79	+	func (f subFixture) open(pendingCount int) atomic.Int32 {
			80	+	released := new(atomic.Int32)
78		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()	81		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()
79	-	f.limiter.EXPECT().Allow(mock.Anything, mock.Anything).Return(true, time.Duration(0)).Maybe()	82	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, string) (bool, time.Duration, func()) {
			83	+	return true, 0, func() { released.Add(1) }
			84	+	}).Maybe()
80		f.pending.EXPECT().Count(mock.Anything).Return(pendingCount, nil).Maybe()	85		f.pending.EXPECT().Count(mock.Anything).Return(pendingCount, nil).Maybe()
			86	+	return released
81		}	87		}
82			88
83		func (f *subFixture) challenge(addr string) ports.IssueChallenge {	89		func (f *subFixture) challenge(addr string) ports.IssueChallenge {

@@ -179,7 +185,6 @@ func TestIssueSubmitSuccess(t *testing.T) {
179		ch := f.challenge(f.addr)	185		ch := f.challenge(f.addr)
180		sub := f.valid(ch)	186		sub := f.valid(ch)
181		f.advance(usecase.MinFillTime)	187		f.advance(usecase.MinFillTime)
182	-	f.limiter.EXPECT().Record(mock.Anything, f.addr).Once()
183		var put domain.PendingIssue	188		var put domain.PendingIssue
184		f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, p domain.PendingIssue) error {	189		f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, p domain.PendingIssue) error {
185		put = p	190		put = p

@@ -217,11 +222,11 @@ func TestIssueSubmitRefusals(t *testing.T) {
217		}},	222		}},
218		{name: "rate limited", want: domain.ErrSubmissionRateLimited, wait: wait, setup: func(f *subFixture) {	223		{name: "rate limited", want: domain.ErrSubmissionRateLimited, wait: wait, setup: func(f *subFixture) {
219		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil)	224		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
220	-	f.limiter.EXPECT().Allow(mock.Anything, mock.Anything).Return(false, 5*time.Minute)	225	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(false, 5*time.Minute, func() {})
221		}},	226		}},
222		{name: "paused at 200", want: domain.ErrSubmissionsPaused, wait: wait, setup: func(f *subFixture) {	227		{name: "paused at 200", want: domain.ErrSubmissionsPaused, wait: wait, setup: func(f *subFixture) {
223		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil)	228		f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
224	-	f.limiter.EXPECT().Allow(mock.Anything, mock.Anything).Return(true, time.Duration(0))	229	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(true, time.Duration(0), func() {})
225		f.pending.EXPECT().Count(mock.Anything).Return(domain.MaxPendingIssues, nil)	230		f.pending.EXPECT().Count(mock.Anything).Return(domain.MaxPendingIssues, nil)
226		}},	231		}},
227		{name: "honeypot", want: domain.ErrSubmissionInvalid, wait: wait, arrange: func(_ subFixture, _ ports.IssueChallenge, s *ports.IssueSubmission) string {	232		{name: "honeypot", want: domain.ErrSubmissionInvalid, wait: wait, arrange: func(_ subFixture, _ ports.IssueChallenge, s *ports.IssueSubmission) string {

@@ -326,7 +331,7 @@ func TestIssueSubmitReplayedNonce(t *testing.T) {
326		ch := f.challenge(f.addr)	331		ch := f.challenge(f.addr)
327		sub := f.valid(ch)	332		sub := f.valid(ch)
328		f.advance(usecase.MinFillTime)	333		f.advance(usecase.MinFillTime)
329	-	// first: text refused, nonce is spent anyway	334	+	// first: text refused, nonce is spent anyway (nothing reserved)
330		bad := sub	335		bad := sub
331		bad.Title = ""	336		bad.Title = ""
332		var te *domain.TextError	337		var te *domain.TextError

@@ -344,7 +349,6 @@ func TestIssueSubmitReplayAfterSuccess(t *testing.T) {
344		ch := f.challenge(f.addr)	349		ch := f.challenge(f.addr)
345		sub := f.valid(ch)	350		sub := f.valid(ch)
346		f.advance(usecase.MinFillTime)	351		f.advance(usecase.MinFillTime)
347	-	f.limiter.EXPECT().Record(mock.Anything, f.addr).Once()
348		f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Once()	352		f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Once()
349		f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Once()	353		f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Once()
350		require.NoError(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub))	354		require.NoError(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub))

@@ -390,7 +394,6 @@ func TestIssueSubmitSpentSetExpiry(t *testing.T) {
390		t.Parallel()	394		t.Parallel()
391		f := newSubFixture(t)	395		f := newSubFixture(t)
392		f.open(0)	396		f.open(0)
393	-	f.limiter.EXPECT().Record(mock.Anything, mock.Anything).Maybe()
394		f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Maybe()	397		f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Maybe()
395		f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()	398		f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()
396		ch := f.challenge(f.addr)	399		ch := f.challenge(f.addr)

@@ -469,3 +472,199 @@ func TestDifficulty(t *testing.T) {
469		assert.Equal(t, 18, usecase.BaseDifficulty)	472		assert.Equal(t, 18, usecase.BaseDifficulty)
470		assert.Equal(t, 21, usecase.MaxDifficulty)	473		assert.Equal(t, 21, usecase.MaxDifficulty)
471		}	474		}
			475	+
			476	+	// submitters builds n valid, distinct submissions (one challenge each) for the
			477	+	// addresses addrOf(i) and the repositories repoOf(i), ready to be sent.
			478	+	type prepared struct {
			479	+	repo domain.RepoName
			480	+	addr string
			481	+	sub ports.IssueSubmission
			482	+	}
			483	+
			484	+	func (f *subFixture) prepare(n int, addrOf func(int) string, repoOf func(int) domain.RepoName) []prepared {
			485	+	out := make([]prepared, n)
			486	+	for i := range n {
			487	+	addr := addrOf(i)
			488	+	repo := repoOf(i)
			489	+	f.limiter.EXPECT().Recent(mock.Anything, addr).Return(0, 0).Once()
			490	+	ch, err := f.uc.Challenge(context.Background(), repo, addr)
			491	+	require.NoError(f.t, err)
			492	+	out[i] = prepared{repo: repo, addr: addr, sub: f.valid(ch)}
			493	+	}
			494	+	f.advance(usecase.MinFillTime)
			495	+	return out
			496	+	}
			497	+
			498	+	// race sends every prepared submission at once and returns the errors.
			499	+	func (f *subFixture) race(ps []prepared) []error {
			500	+	errs := make([]error, len(ps))
			501	+	var wg sync.WaitGroup
			502	+	start := make(chan struct{})
			503	+	for i, p := range ps {
			504	+	wg.Add(1)
			505	+	go func() {
			506	+	defer wg.Done()
			507	+	<-start
			508	+	errs[i] = f.uc.Submit(context.Background(), p.repo, p.addr, p.sub)
			509	+	}()
			510	+	}
			511	+	close(start)
			512	+	wg.Wait()
			513	+	return errs
			514	+	}
			515	+
			516	+	func TestIssueSubmitConcurrentSameAddrRespectsQuota(t *testing.T) {
			517	+	t.Parallel()
			518	+	f := newSubFixture(t)
			519	+	f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()
			520	+	f.pending.EXPECT().Count(mock.Anything).Return(0, nil).Maybe()
			521	+	// a real, atomic limiter semantics: PerHour grants then refusals
			522	+	var granted atomic.Int32
			523	+	var rlMu sync.Mutex
			524	+	f.limiter.EXPECT().Reserve(mock.Anything, f.addr).RunAndReturn(func(context.Context, string) (bool, time.Duration, func()) {
			525	+	rlMu.Lock()
			526	+	defer rlMu.Unlock()
			527	+	if granted.Load() >= 3 {
			528	+	return false, time.Minute, func() {}
			529	+	}
			530	+	granted.Add(1)
			531	+	return true, 0, func() {}
			532	+	}).Maybe()
			533	+	var puts atomic.Int32
			534	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, domain.PendingIssue) error {
			535	+	puts.Add(1)
			536	+	return nil
			537	+	}).Maybe()
			538	+	f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()
			539	+	ps := f.prepare(20, func(int) string { return f.addr }, func(int) domain.RepoName { return f.repo.Name })
			540	+	var ok, limited int
			541	+	for _, err := range f.race(ps) {
			542	+	var re *domain.SubmissionRetryError
			543	+	switch {
			544	+	case err == nil:
			545	+	ok++
			546	+	case errors.As(err, &re):
			547	+	limited++
			548	+	default:
			549	+	t.Fatalf("unexpected error: %v", err)
			550	+	}
			551	+	}
			552	+	assert.Equal(t, 3, ok)
			553	+	assert.Equal(t, 17, limited)
			554	+	assert.Equal(t, int32(3), puts.Load())
			555	+	}
			556	+
			557	+	func TestIssueSubmitConcurrentRespectsPendingCap(t *testing.T) {
			558	+	t.Parallel()
			559	+	f := newSubFixture(t)
			560	+	f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(false, nil).Maybe()
			561	+	f.limiter.EXPECT().Reserve(mock.Anything, mock.Anything).Return(true, time.Duration(0), func() {}).Maybe()
			562	+	other := name(t, "other")
			563	+	f.meta.EXPECT().Get(mock.Anything, other).Return(domain.Repo{Name: other, Visibility: domain.VisibilityPublic, Settings: withIssues(domain.IssuesPublic)}, nil).Maybe()
			564	+	var count atomic.Int32
			565	+	count.Store(domain.MaxPendingIssues - 1)
			566	+	f.pending.EXPECT().Count(mock.Anything).RunAndReturn(func(context.Context) (int, error) {
			567	+	return int(count.Load()), nil
			568	+	}).Maybe()
			569	+	var puts atomic.Int32
			570	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, domain.PendingIssue) error {
			571	+	puts.Add(1)
			572	+	count.Add(1)
			573	+	return nil
			574	+	}).Maybe()
			575	+	f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Maybe()
			576	+	ps := f.prepare(20, func(i int) string { return "ip:198.51.100." + strconv.Itoa(i+1) }, func(i int) domain.RepoName {
			577	+	if i%2 == 0 {
			578	+	return f.repo.Name
			579	+	}
			580	+	return other
			581	+	})
			582	+	var ok, paused int
			583	+	for _, err := range f.race(ps) {
			584	+	switch {
			585	+	case err == nil:
			586	+	ok++
			587	+	case errors.Is(err, domain.ErrSubmissionsPaused):
			588	+	paused++
			589	+	default:
			590	+	t.Fatalf("unexpected error: %v", err)
			591	+	}
			592	+	}
			593	+	assert.Equal(t, 1, ok)
			594	+	assert.Equal(t, 19, paused)
			595	+	assert.Equal(t, int32(1), puts.Load())
			596	+	}
			597	+
			598	+	func TestIssueSubmitReleasesTheReservationWhenStoreFails(t *testing.T) {
			599	+	t.Parallel()
			600	+	f := newSubFixture(t)
			601	+	released := f.open(0)
			602	+	boom := errors.New("disk full")
			603	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(boom).Once()
			604	+	ch := f.challenge(f.addr)
			605	+	sub := f.valid(ch)
			606	+	f.advance(usecase.MinFillTime)
			607	+	require.ErrorIs(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub), boom)
			608	+	assert.Equal(t, int32(1), released.Load())
			609	+
			610	+	f.pending.EXPECT().Put(mock.Anything, mock.Anything).Return(nil).Once()
			611	+	f.bus.EXPECT().Publish(mock.Anything, mock.Anything).Once()
			612	+	ch = f.challenge(f.addr)
			613	+	sub = f.valid(ch)
			614	+	f.advance(usecase.MinFillTime)
			615	+	require.NoError(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub))
			616	+	assert.Equal(t, int32(1), released.Load(), "a stored submission keeps its reservation")
			617	+	}
			618	+
			619	+	func TestIssueSubmitReleasesTheReservationWhenPaused(t *testing.T) {
			620	+	t.Parallel()
			621	+	f := newSubFixture(t)
			622	+	released := f.open(domain.MaxPendingIssues)
			623	+	ch := f.challenge(f.addr)
			624	+	sub := f.valid(ch)
			625	+	f.advance(usecase.MinFillTime)
			626	+	assert.ErrorIs(t, f.uc.Submit(context.Background(), f.repo.Name, f.addr, sub), domain.ErrSubmissionsPaused)
			627	+	assert.Equal(t, int32(1), released.Load())
			628	+	}
			629	+
			630	+	// TestIssueSubmitNoStateOracleBeforeProofOfWork: a banned client, a limited
			631	+	// one, or a full queue all get the same answer to an invalid submission, and
			632	+	// no limiter, count or store is touched (the mocks have no expectation).
			633	+	func TestIssueSubmitNoStateOracleBeforeProofOfWork(t *testing.T) {
			634	+	t.Parallel()
			635	+	f := newSubFixture(t)
			636	+	ch := f.challenge(f.addr)
			637	+	valid := f.valid(ch)
			638	+	f.advance(usecase.MinFillTime)
			639	+	bad := []ports.IssueSubmission{
			640	+	{},
			641	+	{Token: "garbage", Solution: "x", FormToken: "y", Title: "", Body: ""},
			642	+	{Token: ch.Token, Honeypot: "x"},
			643	+	{Token: ch.Token, FormToken: ch.FormToken, Solution: "nope"},
			644	+	{Token: valid.Token, FormToken: valid.FormToken, Solution: "abc", Title: "two\nlines"}, // text error + bad pow
			645	+	}
			646	+	for _, s := range bad {
			647	+	err := f.uc.Submit(context.Background(), f.repo.Name, f.addr, s)
			648	+	require.ErrorIs(t, err, domain.ErrSubmissionInvalid)
			649	+	assert.NotErrorIs(t, err, domain.ErrInvalidText)
			650	+	}
			651	+	// a fully valid submission from a banned client: still invalid, not
			652	+	// limited, paused or a text error, even with bad text.
			653	+	f.bans.EXPECT().Banned(mock.Anything, mock.Anything, mock.Anything).Return(true, nil).Once()
			654	+	bannedBadText := valid
			655	+	bannedBadText.Title = ""
			656	+	err := f.uc.Submit(context.Background(), f.repo.Name, f.addr, bannedBadText)
			657	+	require.ErrorIs(t, err, domain.ErrSubmissionInvalid)
			658	+	assert.NotErrorIs(t, err, domain.ErrInvalidText)
			659	+	var re *domain.SubmissionRetryError
			660	+	assert.False(t, errors.As(err, &re))
			661	+	assert.NotErrorIs(t, err, domain.ErrSubmissionsPaused)
			662	+	}
			663	+
			664	+	func TestIssueChallengeRevealsNoState(t *testing.T) {
			665	+	t.Parallel()
			666	+	f := newSubFixture(t) // no Banned, Reserve or Count expectation
			667	+	f.limiter.EXPECT().Recent(mock.Anything, f.addr).Return(50, 500).Once()
			668	+	_, err := f.uc.Challenge(context.Background(), f.repo.Name, f.addr)
			669	+	require.NoError(t, err)
			670	+	}