feat: enforce configurable per-mailbox campaign admission caps with UTC budgets, accurate diagnostics and PostgreSQL regressions

2026-10-26 00:03:46 +00:00 · 2026-10-09 04:18:52 +00:00
parent 166637b395
commit c6c79d8dcd
8 changed files with 420 additions and 22 deletions
@@ -142,6 +142,20 @@ jobs:
    needs: changes
    if: needs.changes.outputs.go-services == 'true'
    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: warmbly
+          POSTGRES_PASSWORD: warmbly
+          POSTGRES_DB: warmbly_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U warmbly -d warmbly_test"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
    steps:
      - uses: actions/checkout@v4

@@ -176,6 +190,14 @@ jobs:
        run: |
          go test -race -coverprofile=coverage.out -covermode=atomic ./...

+      - name: Campaign send admission regressions (PostgreSQL)
+        env:
+          PRIMARY_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
+          WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
+        run: |
+          go run ./cmd/migrate
+          go test -race ./internal/repository ./internal/tasks -run 'TestLiveCampaignAdmission|TestLiveSharedAdmission|TestLiveDeferredTicksDoNotSpendTheDailyBudget|TestLiveCampaignAtDailyCapWaitsForTomorrow' -count=1
+
      - name: Test the Kafka codec and client lifecycle
        run: go test -race -tags kafka ./internal/infrastructure/codec/ ./internal/infrastructure/kafka/ ./internal/infrastructure/eventbus/

@@ -7,6 +7,8 @@ Campaigns are the cold outreach unit in Warmbly. A campaign holds sending rules,

 All errors follow the shared `{error, message, code, request_id}` envelope documented in [error codes](/api/error-codes/). Authentication and the scope model are covered in [authentication](/api/authentication/) and [permissions](/api/permissions/).

+`daily_limit` applies separately to each mailbox, not to the total sent by the campaign. Campaign send budgets use UTC calendar days; `timezone` controls the sending windows instead. Each mailbox is also constrained by its own cap, other campaigns sharing it, shared daily and rolling recipient limits, and scheduler safety clamps. Dispatch reservations count once, while campaign wake-ups that send no email do not consume the cap. See [campaign sending limits](/guides/campaigns/#sending-limits).
+
 ## List campaigns

 `GET /campaigns`
@@ -120,7 +122,7 @@ Project an audience against a sender pool before a campaign exists. The campaign
 | --- | --- | --- | --- |
 | `segment_ids` | string[] | yes | Segments making up the audience (at most 20). A contact in several of them is counted once. Send `[]` to project the pool alone: its mailboxes and capacity, with no finish date. |
 | `email_tag_ids` | string[] | no | Mailbox tags that resolve the pool. Omit or send `[]` for every active mailbox in the workspace. |
-| `daily_limit` | integer | no | Per-mailbox campaign cap to apply (defaults to `50`). Each mailbox counts the smaller of this and its own cap, lowered further by any clamp the scheduler applies. |
+| `daily_limit` | integer | no | Daily campaign send cap per mailbox, from `3` to `5000` (defaults to `50`). |
 | `days` | integer (0-127) | no | Weekday bitmask of sending days, bit 0 = Monday. Defaults to weekdays. |
 | `timezone` | string | no | IANA timezone the days and window are read in. Omit or send `""` to use the workspace timezone (UTC when none is set). |
 | `start_date` | string (RFC 3339) | no | When sending begins. Omit for now. |
@@ -191,7 +193,7 @@ Create a campaign. Only `name` is required, every other field is optional and ap
 | `utm_medium` | string | no | Overrides the default `email`. Empty means the default. |
 | `utm_campaign` | string | no | Overrides the default, the campaign name as a slug. Empty means the default. |
 | `text_only` | boolean | no | Send plain text only: no HTML part, and open and click tracking are off regardless of their flags. |
-| `daily_limit` | integer | no | Per-campaign daily send cap. |
+| `daily_limit` | integer | no | Daily campaign send cap per mailbox, from `3` to `5000`. |
 | `unsubscribe_header` | boolean | no | Add the RFC 8058 one-click unsubscribe header. |
 | `risky_emails` | boolean | no | Allow sending to risky/unverified addresses. |
 | `cc` | string[] | no | Static CC list, on every email to every lead. An address that is suppressed, is the lead's own, or has bounced on this campaign is left off that email. To copy someone on one lead only, see [set a lead's CC](#set-a-leads-cc). |
@@ -282,6 +282,10 @@ Within a window Warmbly spreads the day's emails evenly, adds random jitter, and

 The spread covers every mailbox on the campaign, not one at a time: the day's remaining capacity across the whole sending pool is paced across the hours left in the window. Adding a second healthy mailbox roughly doubles how fast a campaign works through its leads, while each mailbox still stays inside its own daily cap, hourly ceiling and minimum gap.

+**Daily limit per mailbox** caps this campaign separately for each sender, not across the whole pool. With a limit of `50`, one mailbox sending `50` does not use another mailbox's allowance. Campaign send budgets reset at midnight UTC, independently of the timezone used for sending windows. Sends reserved for dispatch count once; campaign re-checks that send no email do not count. The mailbox's own campaign cap, capacity used by other campaigns, shared daily and rolling recipient limits, and safety clamps still apply. Existing campaigns stored with a non-positive limit keep their uncapped campaign-specific allowance, without bypassing mailbox limits; new API settings remain validated from `3` to `5000`.
+
+If send admission refuses a mailbox at its campaign limit, the activity log reports `CAMPAIGN_DAILY_LIMIT_REACHED`, not a worker outage. A refusal before dispatch gives the lead's reservation back without spending a failed-send attempt; the campaign retries through its normal scheduler.
+
 The spacing sits between sends, never in front of the first one. A campaign you start (or resume) with a lead due sends that lead as soon as one of its mailboxes can (inside the sending window, under its daily cap and past its minimum gap), and the paced gap only begins once that email has gone.

 When nothing is due (every lead is inside a step's wait, for example), the campaign re-checks itself at least every `15` minutes rather than sleeping until the next step falls due; the activity log notes it once a day rather than on every check. Leads added to a waiting campaign are sent on the same terms, without waiting for that re-check or a restart, unless the campaign sets a delay before the first email. Leads added while a campaign is mid-send join the queue and go out at its current pace, so adding leads never shortens the gap between two sends.
@@ -0,0 +1,269 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+type campaignAdmissionFixture struct {
+	*poolLinkFixture
+	r                      *taskRepository
+	worker, campaign, step uuid.UUID
+	historyContact         uuid.UUID
+}
+
+func newCampaignAdmissionFixture(t *testing.T, limit int) *campaignAdmissionFixture {
+	t.Helper()
+	base, r := lineageFixture(t)
+	f := &campaignAdmissionFixture{poolLinkFixture: base, r: r, worker: uuid.New(), campaign: uuid.New(), step: uuid.New()}
+	f.exec(t, `INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol) VALUES($1,'worker',true,NOW(),2)`, f.worker)
+	f.exec(t, `INSERT INTO workers(id) VALUES($1)`, f.worker)
+	t.Cleanup(func() {
+		f.exec(t, `DELETE FROM campaigns WHERE id=$1`, f.campaign)
+		f.exec(t, `DELETE FROM contacts WHERE organization_id=$1`, f.org)
+		f.exec(t, `DELETE FROM fleet_nodes WHERE id=$1`, f.worker)
+	})
+	f.exec(t, `UPDATE email_accounts SET worker_id=$1,campaign_limit=5000,shared_daily_limit=10000,rolling_recipient_limit=10000 WHERE organization_id=$2`, f.worker, f.org)
+	f.exec(t, `INSERT INTO campaigns(id,user_id,organization_id,name,description,status,daily_limit,timezone,days,created_at,updated_at) VALUES($1,$2,$3,'Admission','','active',$4,'Pacific/Honolulu',127,NOW(),NOW())`, f.campaign, f.user, f.org, limit)
+	f.exec(t, `INSERT INTO sequences(id,campaign_id,organization_id,name,subject,body_plain,body_html,wait_after,position) VALUES($1,$2,$3,'Step','Test','Text','',0,1)`, f.step, f.campaign, f.org)
+	f.historyContact = f.contact(t)
+	return f
+}
+
+func (f *campaignAdmissionFixture) exec(t *testing.T, query string, args ...any) {
+	t.Helper()
+	if _, err := f.pool.Exec(context.Background(), query, args...); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *campaignAdmissionFixture) contact(t *testing.T) uuid.UUID {
+	t.Helper()
+	id := uuid.New()
+	f.exec(t, `INSERT INTO contacts(id,user_id,organization_id,email,first_name,last_name,company,phone,custom_fields) VALUES($1,$2,$3,$4,'Test','','','','{}')`, id, f.user, f.org, id.String()+"@example.test")
+	f.exec(t, `INSERT INTO campaign_leads(campaign_id,contact_id,position) VALUES($1,$2,0)`, f.campaign, id)
+	return id
+}
+
+func (f *campaignAdmissionFixture) request(t *testing.T, mailbox uuid.UUID) (OutboundReservation, uuid.UUID) {
+	t.Helper()
+	id, contact := uuid.New(), f.contact(t)
+	f.exec(t, `INSERT INTO tasks(id,task_type,email_account_id,status,message_id) VALUES($1,'campaign',$2,'active','')`, id, mailbox)
+	f.exec(t, `INSERT INTO campaign_tasks(task_id,campaign_id,sequence_id,contact_id) VALUES($1,$2,$3,$4)`, id, f.campaign, f.step, contact)
+	claimed, err := NewCampaignProgressRepository(f.pool).ReserveSend(t.Context(), f.campaign, contact, f.step, id, mailbox, true)
+	if err != nil || !claimed {
+		t.Fatalf("reserve lead: %v %v", claimed, err)
+	}
+	return OutboundReservation{TaskID: id, MailboxID: mailbox, OrganizationID: f.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{contact.String() + "@example.test"}}, contact
+}
+
+func (f *campaignAdmissionFixture) history(t *testing.T, mailbox uuid.UUID, count, days int, message string, wakeup bool) []uuid.UUID {
+	t.Helper()
+	var step any = f.step
+	if wakeup {
+		step = nil
+	}
+	rows, err := f.pool.Query(t.Context(), `WITH history AS (
+	 INSERT INTO tasks(id,task_type,email_account_id,status,message_id,completed_at)
+	 SELECT gen_random_uuid(),'campaign',$1,'completed',$4,NOW()+$3::int*INTERVAL '1 day' FROM generate_series(1,$2::int) RETURNING id)
+	 INSERT INTO campaign_tasks(task_id,campaign_id,sequence_id,contact_id)
+	 SELECT id,$5,$6,$7 FROM history RETURNING task_id`, mailbox, count, days, message, f.campaign, step, f.historyContact)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var ids []uuid.UUID
+	for rows.Next() {
+		var id uuid.UUID
+		if err := rows.Scan(&id); err != nil {
+			t.Fatal(err)
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return ids
+}
+
+func TestLiveCampaignAdmissionPerMailbox(t *testing.T) {
+	f := newCampaignAdmissionFixture(t, 50)
+	ids := f.history(t, f.sender, 50, 0, "<sent@example.test>", false)
+	f.exec(t, `INSERT INTO contacts(id,user_id,organization_id,email,first_name,last_name,company,phone,custom_fields)
+	 SELECT id,$2,$3,id::text||'@example.test','Sent','','','','{}' FROM tasks WHERE id=ANY($1)`, ids, f.user, f.org)
+	f.exec(t, `INSERT INTO campaign_leads(campaign_id,contact_id,position) SELECT $2,id,0 FROM tasks WHERE id=ANY($1)`, ids, f.campaign)
+	f.exec(t, `UPDATE campaign_tasks SET contact_id=task_id WHERE task_id=ANY($1)`, ids)
+	f.exec(t, `INSERT INTO campaign_contact_progress(campaign_id,contact_id,sequence_id,dispatch_task_id,dispatched_at,sent_at)
+	 SELECT $2,id,$3,id,completed_at,completed_at FROM tasks WHERE id=ANY($1)`, ids, f.campaign, f.step)
+	counts, err := f.r.CountCampaignSendsTodayBySender(t.Context(), f.campaign)
+	if err != nil || counts[f.sender] != 50 || counts[f.recipient] != 0 {
+		t.Fatalf("scheduler counts: %v %v", counts, err)
+	}
+	atCap, _ := f.request(t, f.sender)
+	if _, err := f.r.ReserveOutbound(t.Context(), atCap); !errors.Is(err, ErrCampaignDailyLimit) || !errors.Is(err, ErrSendAdmissionDenied) {
+		t.Fatalf("full mailbox: %v", err)
+	}
+	fresh, _ := f.request(t, f.recipient)
+	nonce, err := f.r.ReserveOutbound(t.Context(), fresh)
+	if err != nil {
+		t.Fatalf("unused mailbox after 50 sends on another: %v", err)
+	}
+	state, err := f.r.BeginOutbound(t.Context(), fresh.TaskID, fresh.MailboxID, fresh.WorkerID, nonce)
+	if err != nil || state.State != "execute" {
+		t.Fatalf("second admission: %+v %v", state, err)
+	}
+	counts, err = f.r.CountCampaignSendsTodayBySender(t.Context(), f.campaign)
+	if err != nil || counts[f.sender] != 50 || counts[f.recipient] != 1 {
+		t.Fatalf("reservation counted once: %v %v", counts, err)
+	}
+}
+
+func TestLiveCampaignAdmissionConfiguredLimitsAndProvenance(t *testing.T) {
+	for _, tc := range []struct {
+		name           string
+		limit, sends   int
+		days           int
+		wakeup, denied bool
+		progressOnly   bool
+	}{
+		{name: "configured seven", limit: 7, sends: 7, denied: true},
+		{name: "zero unlimited", limit: 0, sends: 50},
+		{name: "negative unlimited", limit: -1, sends: 50},
+		{name: "previous day", limit: 1, sends: 50, days: -1},
+		{name: "wake ups", limit: 1, sends: 50, wakeup: true},
+		{name: "dispatch provenance without message id", limit: 1, sends: 1, progressOnly: true, denied: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newCampaignAdmissionFixture(t, tc.limit)
+			message := "<legacy@example.test>"
+			if tc.wakeup || tc.progressOnly {
+				message = ""
+			}
+			ids := f.history(t, f.sender, tc.sends, tc.days, message, tc.wakeup)
+			if tc.progressOnly {
+				f.exec(t, `INSERT INTO campaign_contact_progress(campaign_id,contact_id,sequence_id,dispatch_task_id,dispatched_at,sent_at) VALUES($1,$2,$3,$4,NOW(),NOW())`, f.campaign, f.historyContact, f.step, ids[0])
+			}
+			request, _ := f.request(t, f.sender)
+			nonce, err := f.r.ReserveOutbound(t.Context(), request)
+			if tc.denied {
+				if !errors.Is(err, ErrCampaignDailyLimit) {
+					t.Fatalf("want campaign daily limit, got %v", err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			state, err := f.r.BeginOutbound(t.Context(), request.TaskID, request.MailboxID, request.WorkerID, nonce)
+			if err != nil || state.State != "execute" {
+				t.Fatalf("own reservation spent the last slot twice: %+v %v", state, err)
+			}
+		})
+	}
+}
+
+func TestLiveCampaignAdmissionUsesUTCBudget(t *testing.T) {
+	f := newCampaignAdmissionFixture(t, 1)
+	ids := f.history(t, f.sender, 1, 0, "<legacy@example.test>", false)
+	f.exec(t, `UPDATE tasks SET completed_at=(date_trunc('day',NOW() AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') WHERE id=$1`, ids[0])
+	config := f.pool.Config()
+	config.ConnConfig.RuntimeParams["timezone"] = "Pacific/Honolulu"
+	pool, err := pgxpool.NewWithConfig(t.Context(), config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	r := NewTaskRepository(pool).(*taskRepository)
+	counts, err := r.CountCampaignSendsTodayBySender(t.Context(), f.campaign)
+	if err != nil || counts[f.sender] != 1 {
+		t.Fatalf("scheduler UTC count: %v %v", counts, err)
+	}
+	request, _ := f.request(t, f.sender)
+	if _, err := r.ReserveOutbound(t.Context(), request); !errors.Is(err, ErrCampaignDailyLimit) {
+		t.Fatalf("admission differs from scheduler in non-UTC session: %v", err)
+	}
+}
+
+func TestLiveCampaignAdmissionPreservesMailboxControls(t *testing.T) {
+	for _, column := range []string{"campaign_limit", "shared_daily_limit", "rolling_recipient_limit"} {
+		t.Run(column, func(t *testing.T) {
+			f := newCampaignAdmissionFixture(t, 0)
+			f.history(t, f.sender, 1, 0, "<legacy@example.test>", false)
+			f.exec(t, `UPDATE email_accounts SET `+column+`=1 WHERE id=$1`, f.sender)
+			request, _ := f.request(t, f.sender)
+			if _, err := f.r.ReserveOutbound(t.Context(), request); !errors.Is(err, ErrSendAdmissionDenied) || errors.Is(err, ErrCampaignDailyLimit) {
+				t.Fatalf("mailbox control bypassed or mislabeled: %v", err)
+			}
+		})
+	}
+}
+
+func TestLiveCampaignAdmissionConcurrentLastSlotAndCancellation(t *testing.T) {
+	f := newCampaignAdmissionFixture(t, 7)
+	f.history(t, f.sender, 6, 0, "<legacy@example.test>", false)
+	requests := make([]OutboundReservation, 12)
+	for i := range requests {
+		requests[i], _ = f.request(t, f.sender)
+	}
+	start := make(chan struct{})
+	wins := make(chan OutboundReservation, len(requests))
+	errCh := make(chan error, len(requests))
+	var wg sync.WaitGroup
+	for _, request := range requests {
+		wg.Go(func() {
+			<-start
+			if _, err := f.r.ReserveOutbound(t.Context(), request); err == nil {
+				wins <- request
+			} else if !errors.Is(err, ErrSendAdmissionDenied) {
+				errCh <- err
+			}
+		})
+	}
+	close(start)
+	wg.Wait()
+	close(wins)
+	close(errCh)
+	for err := range errCh {
+		t.Fatal(err)
+	}
+	if len(wins) != 1 {
+		t.Fatalf("admitted %d sends into one slot", len(wins))
+	}
+	winner := <-wins
+	var nonce uuid.UUID
+	if err := f.pool.QueryRow(t.Context(), `SELECT send_executor_nonce FROM tasks WHERE id=$1`, winner.TaskID).Scan(&nonce); err != nil {
+		t.Fatal(err)
+	}
+	// A changed cap is rechecked before any command executes.
+	f.exec(t, `UPDATE campaigns SET daily_limit=6 WHERE id=$1`, f.campaign)
+	state, err := f.r.BeginOutbound(t.Context(), winner.TaskID, winner.MailboxID, winner.WorkerID, nonce)
+	if err != nil || state.State != "denied" {
+		t.Fatalf("worker did not recheck campaign capacity: %+v %v", state, err)
+	}
+	if err := f.r.CancelOutbound(t.Context(), winner.TaskID, winner.MailboxID, winner.WorkerID, nonce); err != nil {
+		t.Fatal(err)
+	}
+	var reason, code string
+	if err := f.pool.QueryRow(t.Context(), `SELECT message FROM task_failures WHERE task_id=$1`, winner.TaskID).Scan(&reason); err != nil || reason != ErrCampaignDailyLimit.Error() {
+		t.Fatalf("worker denial lost its task failure reason: %s %v", reason, err)
+	}
+	if err := f.pool.QueryRow(t.Context(), `SELECT metadata->>'code' FROM campaign_logs WHERE campaign_id=$1 AND metadata->>'task_id'=$2`, f.campaign, winner.TaskID.String()).Scan(&code); err != nil || code != "CAMPAIGN_DAILY_LIMIT_REACHED" {
+		t.Fatalf("worker denial lost its capacity diagnostic: %s %v", code, err)
+	}
+	counts, err := f.r.CountCampaignSendsTodayBySender(t.Context(), f.campaign)
+	if err != nil || counts[f.sender] != 6 {
+		t.Fatalf("cancelled reservation still consumes the cap: %v %v", counts, err)
+	}
+	for _, request := range requests {
+		var reserved, held bool
+		if err := f.pool.QueryRow(t.Context(), `SELECT send_reserved_at IS NOT NULL AND send_released_at IS NULL,ea.send_recovery_hold FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, request.TaskID).Scan(&reserved, &held); err != nil || reserved || held {
+			t.Fatalf("reservation leaked after cancellation: %v %v %v", reserved, held, err)
+		}
+	}
+}
@@ -465,6 +465,10 @@ func (r *taskRepository) GetEmailTask(ctx context.Context, taskID uuid.UUID) (*E
 const taskDispatchedEmail = `(t.task_type <> 'campaign' OR t.message_id <> '' OR EXISTS (
 		SELECT 1 FROM campaign_contact_progress ccp WHERE ccp.dispatch_task_id = t.id))`

+const campaignSendToday = `t.status = 'completed' AND t.task_type = 'campaign'
+	AND (t.completed_at AT TIME ZONE 'UTC')::date = (NOW() AT TIME ZONE 'UTC')::date
+	AND ` + taskDispatchedEmail
+
 // CountCampaignEmailsSentToday counts the campaign emails a mailbox dispatched
 // today (excludes warmup, and the campaign chain's own wake-ups). Placement
 // test probes count too: they are cold mail from the same daily budget.
@@ -476,10 +480,7 @@ func (r *taskRepository) CountCampaignEmailsSentToday(ctx context.Context, accou
 			SELECT COUNT(*)
 			FROM tasks t
 			WHERE t.email_account_id = $1
-			  AND t.status = 'completed'
-			  AND t.task_type = 'campaign'
-			  AND DATE(t.completed_at) = CURRENT_DATE
-			  AND ` + taskDispatchedEmail + `
+			  AND ` + campaignSendToday + `
 		) + (` + placementSentTodaySQL + ` AND t.email_account_id = $1)
 	`

@@ -530,10 +531,7 @@ func (r *taskRepository) CountCampaignEmailsSentTodayByAccounts(ctx context.Cont
 		SELECT t.email_account_id, COUNT(*)
 		FROM tasks t
 		WHERE t.email_account_id = ANY($1)
-		  AND t.status = 'completed'
-		  AND t.task_type = 'campaign'
-		  AND DATE(t.completed_at) = CURRENT_DATE
-		  AND ` + taskDispatchedEmail + `
+		  AND ` + campaignSendToday + `
 		GROUP BY t.email_account_id
 		UNION ALL
 		SELECT t.email_account_id, COUNT(*)
@@ -612,10 +610,7 @@ func (r *taskRepository) CountCampaignSendsTodayBySender(ctx context.Context, ca
 		FROM tasks t
 		JOIN campaign_tasks ct ON ct.task_id = t.id
 		WHERE ct.campaign_id = $1
-		  AND t.status = 'completed'
-		  AND t.task_type = 'campaign'
-		  AND DATE(t.completed_at) = CURRENT_DATE
-		  AND ` + taskDispatchedEmail + `
+		  AND ` + campaignSendToday + `
 		GROUP BY t.email_account_id
 	`
 	rows, err := r.db.Query(ctx, query, campaignID)
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/mail"
 	"sort"
 	"strings"
@@ -17,6 +18,19 @@ import (

 var ErrSendAdmissionDenied = errors.New("current send authority or capacity unavailable")

+var ErrCampaignDailyLimit = fmt.Errorf("%w: campaign daily limit reached for this mailbox", ErrSendAdmissionDenied)
+
+func SendAdmissionCode(err error) string {
+	switch {
+	case errors.Is(err, ErrCampaignDailyLimit):
+		return "CAMPAIGN_DAILY_LIMIT_REACHED"
+	case errors.Is(err, ErrSendAdmissionDenied):
+		return "SEND_ADMISSION_DENIED"
+	default:
+		return ""
+	}
+}
+
 type OutboundReservation struct {
 	TaskID         uuid.UUID
 	MailboxID      uuid.UUID
@@ -241,21 +255,24 @@ func checkCampaignReservation(ctx context.Context, tx pgx.Tx, task uuid.UUID) er
 	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,272))`, campaign.String()); err != nil {
 		return err
 	}
-	var allowed bool
+	var allowed, capacity bool
 	err := tx.QueryRow(ctx, `SELECT c.status='active' AND (c.end_date IS NULL OR c.end_date>=NOW())
 	 AND NOT recipient_suppressed(c.organization_id,contact.email) AND contact.subscribed IS DISTINCT FROM false
 	 AND NOT EXISTS(SELECT 1 FROM campaign_leads h WHERE h.campaign_id=c.id AND h.contact_id=ct.contact_id AND h.paused_at IS NOT NULL AND (h.paused_until IS NULL OR h.paused_until>NOW()))
-	 AND (c.daily_limit<=0 OR (SELECT COUNT(*) FROM campaign_contact_progress p WHERE p.campaign_id=c.id
-	 AND COALESCE(p.sent_at,p.dispatched_at)>=date_trunc('day',NOW() AT TIME ZONE COALESCE(NULLIF(c.timezone,''),NULLIF(o.timezone,''),'UTC')) AT TIME ZONE COALESCE(NULLIF(c.timezone,''),NULLIF(o.timezone,''),'UTC'))<=c.daily_limit)
 	 AND c.organization_id=ea.organization_id AND contact.organization_id=c.organization_id
-	 AND EXISTS(SELECT 1 FROM campaign_contact_progress p WHERE p.campaign_id=c.id AND p.contact_id=ct.contact_id AND p.sequence_id=ct.sequence_id AND p.dispatch_task_id=t.id)
-	 FROM campaign_tasks ct JOIN tasks t ON t.id=ct.task_id JOIN email_accounts ea ON ea.id=t.email_account_id JOIN campaigns c ON c.id=ct.campaign_id JOIN organizations o ON o.id=c.organization_id JOIN contacts contact ON contact.id=ct.contact_id WHERE ct.task_id=$1`, task).Scan(&allowed)
+	 AND EXISTS(SELECT 1 FROM campaign_contact_progress p WHERE p.campaign_id=c.id AND p.contact_id=ct.contact_id AND p.sequence_id=ct.sequence_id AND p.dispatch_task_id=t.id),
+	 c.daily_limit<=0 OR (SELECT COUNT(*) FROM tasks t JOIN campaign_tasks sent ON sent.task_id=t.id
+	 WHERE sent.campaign_id=c.id AND t.email_account_id=ea.id AND t.id<>ct.task_id AND `+campaignSendToday+`)<c.daily_limit
+	 FROM campaign_tasks ct JOIN tasks t ON t.id=ct.task_id JOIN email_accounts ea ON ea.id=t.email_account_id JOIN campaigns c ON c.id=ct.campaign_id JOIN contacts contact ON contact.id=ct.contact_id WHERE ct.task_id=$1`, task).Scan(&allowed, &capacity)
 	if err != nil {
 		return err
 	}
 	if !allowed {
 		return ErrSendAdmissionDenied
 	}
+	if !capacity {
+		return ErrCampaignDailyLimit
+	}
 	return nil
 }

@@ -332,10 +349,11 @@ func (r *taskRepository) BeginOutbound(ctx context.Context, task, mailbox, worke
 		if !errors.Is(err, ErrSendAdmissionDenied) {
 			return nil, err
 		}
+		reason := err
 		if err = tx.Commit(ctx); err != nil {
 			return nil, err
 		}
-		if err = r.CancelOutbound(ctx, task, mailbox, worker, nonce); err != nil {
+		if err = r.cancelOutbound(ctx, task, mailbox, worker, nonce, reason); err != nil {
 			return nil, err
 		}
 		return &WarmupDispatchState{State: "denied"}, nil
@@ -400,6 +418,10 @@ func (r *taskRepository) FinishOutbound(ctx context.Context, task, mailbox, work
 }

 func (r *taskRepository) CancelOutbound(ctx context.Context, task, mailbox, worker, nonce uuid.UUID) error {
+	return r.cancelOutbound(ctx, task, mailbox, worker, nonce, nil)
+}
+
+func (r *taskRepository) cancelOutbound(ctx context.Context, task, mailbox, worker, nonce uuid.UUID, reason error) error {
 	tx, err := r.db.Begin(ctx)
 	if err != nil {
 		return err
@@ -428,6 +450,16 @@ func (r *taskRepository) CancelOutbound(ctx context.Context, task, mailbox, work
 		if err = NewCampaignProgressRepository(r.db).ReleaseSend(inner, c, contact, sequence, newLead); err != nil {
 			return err
 		}
+		if reason != nil {
+			if _, err = tx.Exec(ctx, `INSERT INTO task_failures(task_id,title,message) VALUES($1,'Send admission denied',$2)
+			 ON CONFLICT(task_id) DO UPDATE SET title=EXCLUDED.title,message=EXCLUDED.message`, task, reason.Error()); err != nil {
+				return err
+			}
+			metadata := map[string]any{"level": "error", "code": SendAdmissionCode(reason), "contact_id": contact.String(), "sequence_id": sequence.String(), "account_id": mailbox.String(), "task_id": task.String(), "error": reason.Error(), "will_retry": true}
+			if _, err = tx.Exec(ctx, `INSERT INTO campaign_logs(campaign_id,event_type,message,metadata) VALUES($1,'email_failed',$2,$3)`, c, "Send admission denied before worker execution, will retry: "+reason.Error(), metadata); err != nil {
+				return err
+			}
+		}
 	}
 	if lane == "warmup" {
 		if _, err = tx.Exec(ctx, `UPDATE warmup_statistics SET emails_sent=GREATEST(emails_sent-1,0),emails_replied=GREATEST(emails_replied-CASE WHEN w.parent_task_id IS NULL THEN 0 ELSE 1 END,0) FROM warmup_tasks w JOIN tasks t ON t.id=w.task_id WHERE w.task_id=$2 AND warmup_statistics.email_account_id=$1 AND date=DATE(t.completed_at)`, mailbox, task); err != nil {
@@ -2,9 +2,67 @@ package tasks

 import (
 	"context"
+	"errors"
+	"fmt"
+	"strings"
 	"testing"
+
+	"github.com/warmbly/warmbly/internal/repository"
 )

+func TestCampaignDispatchFailureCode(t *testing.T) {
+	for _, tc := range []struct {
+		err  error
+		code string
+	}{
+		{repository.ErrCampaignDailyLimit, "CAMPAIGN_DAILY_LIMIT_REACHED"},
+		{repository.ErrSendAdmissionDenied, "SEND_ADMISSION_DENIED"},
+		{ErrWorkerOffline, "WORKER_UNAVAILABLE"},
+		{ErrWorkerUnconfirmed, "WORKER_UNAVAILABLE"},
+		{ErrSendDispatchUnknown, "SEND_DISPATCH_UNKNOWN"},
+		{errors.New("publish failed"), "SEND_FAILED"},
+	} {
+		t.Run(tc.code+"/"+tc.err.Error(), func(t *testing.T) {
+			if got := campaignDispatchFailureCode(fmt.Errorf("dispatch: %w", tc.err)); got != tc.code {
+				t.Fatalf("got %s, want %s", got, tc.code)
+			}
+		})
+	}
+}
+
+func TestLiveCampaignAdmissionFailureReleasesReservationWithoutAnAttempt(t *testing.T) {
+	for _, denied := range []error{repository.ErrCampaignDailyLimit, repository.ErrSendAdmissionDenied} {
+		t.Run(denied.Error(), func(t *testing.T) {
+			f := newSendFixture(t)
+			f.sender.fail = fmt.Errorf("dispatch: %w", denied)
+			id := f.tick(t)
+			var code, message string
+			if err := f.pool.QueryRow(t.Context(), `SELECT metadata->>'code' FROM campaign_logs WHERE campaign_id=$1 AND event_type='email_failed' ORDER BY created_at DESC LIMIT 1`, f.campaign).Scan(&code); err != nil || code != campaignDispatchFailureCode(denied) {
+				t.Fatalf("admission mislabeled as a worker outage: %s %v", code, err)
+			}
+			if err := f.pool.QueryRow(t.Context(), `SELECT message FROM task_failures WHERE task_id=$1`, id).Scan(&message); err != nil || !strings.Contains(message, denied.Error()) {
+				t.Fatalf("task failure lost admission reason: %s %v", message, err)
+			}
+			var reserved, bound bool
+			var attempts, sends, leads int
+			if err := f.pool.QueryRow(t.Context(), `SELECT dispatched_at IS NOT NULL OR dispatch_task_id IS NOT NULL,send_attempts FROM campaign_contact_progress WHERE campaign_id=$1 AND contact_id=$2 AND sequence_id=$3`, f.campaign, f.leadA, f.step).Scan(&reserved, &attempts); err != nil || reserved || attempts != 0 {
+				t.Fatalf("denied handoff kept reservation or spent an attempt: %v %d %v", reserved, attempts, err)
+			}
+			if err := f.pool.QueryRow(t.Context(), `SELECT email_account_id IS NOT NULL FROM campaign_leads WHERE campaign_id=$1 AND contact_id=$2`, f.campaign, f.leadA).Scan(&bound); err != nil || bound {
+				t.Fatalf("denied handoff kept mailbox binding: %v %v", bound, err)
+			}
+			if err := f.pool.QueryRow(t.Context(), `SELECT emails_sent,new_leads_started FROM campaign_daily_sends WHERE campaign_id=$1 AND send_date=CURRENT_DATE`, f.campaign).Scan(&sends, &leads); err != nil || sends != 0 || leads != 0 {
+				t.Fatalf("denied handoff spent a daily reservation: %d %d %v", sends, leads, err)
+			}
+			f.sender.fail = nil
+			f.tick(t)
+			if f.sender.count() != 1 {
+				t.Fatal("released lead could not be retried")
+			}
+		})
+	}
+}
+
 // End-to-end checks for issue #306 over the real campaign tick: the chain's
 // own wake-ups (a deferral, a pause) must not spend the mailbox's daily
 // budget, and a campaign whose mailboxes are all at their cap waits for
@@ -915,15 +915,16 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx
 				}
 			}
 		}
+		failureCode := campaignDispatchFailureCode(err)
 		s.taskRepo.RecordTaskFailure(ctx, taskID, "Send failed", err.Error())
 		if s.campaignLogRepo != nil {
 			s.campaignLogRepo.CreateLog(ctx, &repository.CampaignLogEntry{
 				CampaignID: campaign.ID,
 				EventType:  "email_failed",
-				Message:    fmt.Sprintf("Could not hand %s's email to a sending worker, will retry: %s", contact.Email, err.Error()),
+				Message:    fmt.Sprintf("Could not dispatch %s's email, will retry: %s", contact.Email, err.Error()),
 				Metadata: map[string]interface{}{
 					"level":       "error",
-					"code":        "WORKER_UNAVAILABLE",
+					"code":        failureCode,
 					"contact_id":  contact.ID.String(),
 					"sequence_id": sequence.ID.String(),
 					"account_id":  account.ID.String(),
@@ -938,6 +939,7 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx
 				"campaign_id": campaign.ID.String(),
 				"contact_id":  contact.ID.String(),
 				"error":       err.Error(),
+				"code":        failureCode,
 			})

 			s.streamingPublisher.PublishTaskProgress(ctx, s.sendProgress(ctx, campaign, taskID, contact, sequence, "failed", processedCount, totalEmails, totalContacts))
@@ -1080,6 +1082,20 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx
 	return nil
 }

+func campaignDispatchFailureCode(err error) string {
+	if code := repository.SendAdmissionCode(err); code != "" {
+		return code
+	}
+	switch {
+	case errors.Is(err, ErrWorkerOffline), errors.Is(err, ErrWorkerUnconfirmed):
+		return "WORKER_UNAVAILABLE"
+	case errors.Is(err, ErrSendDispatchUnknown):
+		return "SEND_DISPATCH_UNKNOWN"
+	default:
+		return "SEND_FAILED"
+	}
+}
+
 // threadParent resolves the email this step should be sent as a reply to, or
 // nil when it must open a new conversation: the step has reply-in-thread
 // turned off, the contact has had nothing from this campaign yet, or the