From f63f459c7319db80dbd696303c13db6a42c756ee Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 15 Oct 2026 14:59:09 +0200 Subject: [PATCH] feat: bound unibox sends across a workspace's mailboxes by a fair-use allowance of twice its plan's daily email limit with a floor of 100, resolved through the feature gate so overrides apply and self-hosted or unlimited plans stay unbounded --- docs/content/docs/api/error-codes.mdx | 2 +- docs/content/docs/development/send-safety.mdx | 2 +- docs/content/docs/guides/unibox.mdx | 2 + internal/models/manual_send.go | 21 ++++++++- internal/models/manual_send_test.go | 20 +++++++++ internal/repository/send_admission.go | 30 +++++++++++++ .../repository/shared_admission_live_test.go | 25 +++++++++++ internal/tasks/email_sender.go | 5 ++- .../tasks/user_email_refusal_live_test.go | 44 +++++++++++++++++++ internal/tasks/user_email_task.go | 20 +++++++++ 10 files changed, 167 insertions(+), 4 deletions(-) diff --git a/docs/content/docs/api/error-codes.mdx b/docs/content/docs/api/error-codes.mdx index c5adc1e63..04ef70105 100644 --- a/docs/content/docs/api/error-codes.mdx +++ b/docs/content/docs/api/error-codes.mdx @@ -30,7 +30,7 @@ These codes appear in campaign activity and task failure metadata, not as HTTP r |------|---------| | `CAMPAIGN_DAILY_LIMIT_REACHED` | This campaign has reached its configured UTC-day maximum for this sending mailbox. Other mailboxes have separate campaign allowances. | | `MAILBOX_SENDING_PLAN_DENIED` | The mailbox's enabled local-day sending plan is unavailable, outside working hours or on a break, or its daily/hourly quota is exhausted. The scheduler retries when capacity opens. | -| `MANUAL_SEND_LIMIT_REACHED` | A unibox send (reply, forward or compose, including `POST /unibox/*` and `POST /emails/:id/send`) would exceed the mailbox's fair-use allowance: recipients on one email, in an hour, in 24 hours, or in new conversations over 24 hours. The failure message names which. Unibox sends never count toward campaign or warmup limits. See [sending controls](/guides/unibox/#sending-controls). | +| `MANUAL_SEND_LIMIT_REACHED` | A unibox send (reply, forward or compose, including `POST /unibox/*` and `POST /emails/:id/send`) would exceed the mailbox's fair-use allowance: recipients on one email, in an hour, in 24 hours, in new conversations over 24 hours, or, across all the workspace's mailboxes, the 24-hour allowance its plan gives. The failure message names which. Unibox sends never count toward campaign or warmup limits. See [sending controls](/guides/unibox/#sending-controls). | | `SEND_ADMISSION_DENIED` | Another current send-authority, recipient or mailbox capacity check denied admission. | Admission denial before dispatch releases the send reservation without spending a failed-send attempt. A refused unibox send ends with status `failed`, is not retried, and is reported to the mailbox owner. See [campaigns](/guides/campaigns/) and [sending behavior](/guides/sending-behavior/) for the separate configurable ceilings and daily targets. diff --git a/docs/content/docs/development/send-safety.mdx b/docs/content/docs/development/send-safety.mdx index b6051afcd..31531387c 100644 --- a/docs/content/docs/development/send-safety.mdx +++ b/docs/content/docs/development/send-safety.mdx @@ -31,7 +31,7 @@ Local participation restricts local use of an alias. Individually removing a man ## Shared send-time admission -Every new outbound lane goes through a mailbox/task advisory lock and one durable SQL reservation: campaigns, manual/Unibox sends, normal warmup, campaign-backed health checks/reply-back, and placement/seed probes. Campaign progress already reserved for the same task is not counted twice. Cold and diagnostic allocations remain distinct constraints, with optional shared and rolling ceilings layered above them. Manual Unibox sends (reply, forward, compose) are conversations rather than outreach: they spend none of those allocations and are not refused by them, so a mailbox at its campaign limit can still answer a lead. They are held to fair use instead: a rolling 24-hour recipient allowance sized below the mailbox's provider cap and lowered by connection age, warmup standing and workspace risk, with an hourly, a per-message and a new-conversation bound under it (see [unibox sending controls](/guides/unibox/#sending-controls)). The same mailbox standing as every other lane still applies: active, current worker, no recovery hold or cooldown, no suppressed recipient. Calendar days use the mailbox/organization timezone, including daylight-saving changes; rolling recipient totals do not reset at midnight. +Every new outbound lane goes through a mailbox/task advisory lock and one durable SQL reservation: campaigns, manual/Unibox sends, normal warmup, campaign-backed health checks/reply-back, and placement/seed probes. Campaign progress already reserved for the same task is not counted twice. Cold and diagnostic allocations remain distinct constraints, with optional shared and rolling ceilings layered above them. Manual Unibox sends (reply, forward, compose) are conversations rather than outreach: they spend none of those allocations and are not refused by them, so a mailbox at its campaign limit can still answer a lead. They are held to fair use instead: a rolling 24-hour recipient allowance sized below the mailbox's provider cap and lowered by connection age, warmup standing and workspace risk, with an hourly, a per-message and a new-conversation bound under it, and a workspace-wide allowance of twice the plan's daily email limit (floor 100) checked under a per-workspace lock. `BILLING_PROVIDER=none` and plans without a daily limit set no workspace allowance (see [unibox sending controls](/guides/unibox/#sending-controls)). The same mailbox standing as every other lane still applies: active, current worker, no recovery hold or cooldown, no suppressed recipient. Calendar days use the mailbox/organization timezone, including daylight-saving changes; rolling recipient totals do not reset at midnight. Reservation requires an active organization-owned mailbox, its current configured provider and worker, current suppression and recovery state, and the exact recipient envelope. Campaign sends additionally require active campaign/lead/progress state and current calendar/pacing policy. Diagnostic partners retain their real pool, role, tier, trust, standing and inbound capacity constraints. Seed mailboxes stay measurements, not conversation partners. Future, missing or stale positive Cloud standing is unavailable evidence; restrictive holds are not erased. diff --git a/docs/content/docs/guides/unibox.mdx b/docs/content/docs/guides/unibox.mdx index c6c2f47ac..212481769 100644 --- a/docs/content/docs/guides/unibox.mdx +++ b/docs/content/docs/guides/unibox.mdx @@ -406,6 +406,8 @@ However low that goes, every mailbox keeps at least 50 recipients a day, so a co - one email goes to at most 100 recipients, or the hourly allowance if that is smaller. A longer list belongs in a campaign - new emails that start a conversation take at most a quarter of it. Replies and forwards inside a conversation do not count toward this part, so a mailbox can always keep answering +Your plan sets a second, workspace-wide allowance across all your mailboxes together: twice the plan's daily email limit in any 24 hours, and never less than 100. The free trial's 20 emails a day gives 100 unibox recipients, Starter's 150 gives 300, and Grow's 3,000 gives 6,000. An approved increase to your daily limit raises it too. A plan with no daily limit has no workspace allowance, and neither does a self-hosted instance without billing. + A mailbox that is paused or reconnecting, held after a send problem, or cooling down after a provider limit sends nothing from the unibox either, and neither does a send to an address on your suppression list. When a send is refused for one of those reasons, it is not sent and the mailbox owner sees an error naming the cause and the limit it reached, so they can send it later or from another mailbox. If Warmbly cannot confirm the email reached the mailbox's sending worker, the error says so; check **Sent** before sending it again. diff --git a/internal/models/manual_send.go b/internal/models/manual_send.go index 8cb87e745..eb5e7e15a 100644 --- a/internal/models/manual_send.go +++ b/internal/models/manual_send.go @@ -11,7 +11,7 @@ import ( // are held to fair use instead: a generous daily recipient allowance sized to // what the mailbox's provider accepts, lowered by what is known about the // mailbox and its workspace, with a burst, a new-conversation and a -// per-message bound under it. Replies and forwards inside a conversation only +// per-message bound under it, and a workspace-wide bound from its plan. Replies and forwards inside a conversation only // spend the daily and hourly allowance; new conversations are the cold-outreach // path, so they also spend their own, smaller one. const ( @@ -28,6 +28,11 @@ const ( manualSendNewConvDivisor = 4 // a quarter of the day to new conversations manualSendYoungMailboxDays = 14 manualSendNewMailboxDays = 2 + + // The workspace's plan bounds its unibox sends across every mailbox: + // twice the plan's daily email limit, never under the floor. + ManualSendPlanMultiple = 2 + ManualSendWorkspaceFloor = 100 ) var ( @@ -105,6 +110,16 @@ func manualSendDailyBase(provider InboxProvider, address string) int { } } +// ManualSendWorkspaceAllowance is a workspace's unibox recipients in 24 hours +// across all its mailboxes, from its plan's daily email limit. A negative +// limit is an unlimited plan (or a self-hosted instance) and returns 0, no bound. +func ManualSendWorkspaceAllowance(planDaily int) int { + if planDaily < 0 { + return 0 + } + return max(ManualSendWorkspaceFloor, planDaily*ManualSendPlanMultiple) +} + // ManualSendWindow names the bound a manual send ran into. type ManualSendWindow string @@ -113,6 +128,8 @@ const ( ManualSendWindowHour ManualSendWindow = "hour" ManualSendWindowDay ManualSendWindow = "day" ManualSendWindowNew ManualSendWindow = "new_conversations" + // ManualSendWindowPlan is the workspace's plan allowance, across mailboxes. + ManualSendWindowPlan ManualSendWindow = "plan" ) // ManualSendUsage is what the mailbox has already sent manually, in recipient @@ -145,6 +162,8 @@ func ManualSendRefusal(window ManualSendWindow, limit int) string { return fmt.Sprintf("the mailbox has reached its fair-use limit of %d recipients in an hour. Try again later or send from another mailbox.", limit) case ManualSendWindowNew: return fmt.Sprintf("the mailbox has started conversations with %d recipients in the last 24 hours, its fair-use limit for new emails. Replies still go out; use a campaign for outreach.", limit) + case ManualSendWindowPlan: + return fmt.Sprintf("your workspace has reached its plan's fair-use limit of %d unibox recipients in 24 hours across all mailboxes. Try again later or upgrade the plan for a higher limit.", limit) default: return fmt.Sprintf("the mailbox has reached its fair-use limit of %d recipients in 24 hours. Try again later or send from another mailbox.", limit) } diff --git a/internal/models/manual_send_test.go b/internal/models/manual_send_test.go index b70b8ca9e..9195d6869 100644 --- a/internal/models/manual_send_test.go +++ b/internal/models/manual_send_test.go @@ -65,3 +65,23 @@ func TestManualSendExceededChargesNewConversationsOnly(t *testing.T) { }) } } + +func TestManualSendWorkspaceAllowanceFollowsThePlan(t *testing.T) { + for _, tc := range []struct { + name string + planDaily int + want int + }{ + {"unlimited plan or self-hosted", -1, 0}, + {"free trial (20 a day) keeps the floor", 20, 100}, + {"starter (150 a day)", 150, 300}, + {"grow (3,000 a day)", 3000, 6000}, + {"no daily limit at all still keeps the floor", 0, 100}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := ManualSendWorkspaceAllowance(tc.planDaily); got != tc.want { + t.Errorf("allowance = %d, want %d", got, tc.want) + } + }) + } +} diff --git a/internal/repository/send_admission.go b/internal/repository/send_admission.go index 6fb6dc117..58a7d42a0 100644 --- a/internal/repository/send_admission.go +++ b/internal/repository/send_admission.go @@ -58,6 +58,9 @@ type OutboundReservation struct { WorkerID uuid.UUID Provider models.InboxProvider Recipients []string + // ManualPlanAllowance bounds the workspace's manual (unibox) recipients + // in 24 hours across its mailboxes; 0 means no plan bound. + ManualPlanAllowance int } type OutboundAdmissionRepository interface { @@ -178,6 +181,9 @@ func (r *taskRepository) ReserveOutbound(ctx context.Context, in OutboundReserva if err = checkManualSendFairUse(ctx, tx, in.TaskID, in.MailboxID, len(recipients)); err != nil { return uuid.Nil, err } + if err = checkManualSendPlan(ctx, tx, in.TaskID, in.OrganizationID, len(recipients), in.ManualPlanAllowance); err != nil { + return uuid.Nil, err + } } // Calendar counters and rolling recipient occurrences are independent constraints. var attemptedRecipients, attemptedToday int @@ -303,6 +309,30 @@ func checkManualSendFairUse(ctx context.Context, tx pgx.Tx, task, mailbox uuid.U return nil } +// checkManualSendPlan holds a workspace's manual sends, across all its +// mailboxes, to the allowance its plan gives. The workspace lock keeps two +// mailboxes from both taking the last of it. +func checkManualSendPlan(ctx context.Context, tx pgx.Tx, task, org uuid.UUID, n, allowance int) error { + if allowance <= 0 { + return nil + } + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,273))`, org.String()); err != nil { + return err + } + var used int + err := tx.QueryRow(ctx, `SELECT COALESCE(SUM(GREATEST(1,cardinality(t.send_recipients))),0) + FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id + WHERE ea.organization_id=$1 AND t.id<>$2 AND t.task_type='email' + AND t.send_reserved_at>NOW()-INTERVAL '24 hours' AND t.send_released_at IS NULL`, org, task).Scan(&used) + if err != nil { + return err + } + if used+n > allowance { + return &ManualSendLimitError{Window: models.ManualSendWindowPlan, Limit: allowance} + } + return nil +} + func checkSendRecipients(ctx context.Context, tx pgx.Tx, org uuid.UUID, recipients []string) error { var denied bool err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM unnest($2::text[]) recipient WHERE recipient_suppressed($1,recipient))`, org, recipients).Scan(&denied) diff --git a/internal/repository/shared_admission_live_test.go b/internal/repository/shared_admission_live_test.go index 55a420626..d1b8db19d 100644 --- a/internal/repository/shared_admission_live_test.go +++ b/internal/repository/shared_admission_live_test.go @@ -234,6 +234,31 @@ func TestLiveManualSendIsOutsideTheCappedAllocations(t *testing.T) { if err := send(reply); err != nil { t.Fatalf("threaded reply refused by the new-conversation allowance: %v", err) } + // The plan bounds the workspace across mailboxes: another mailbox's 5 + // recipients count toward this one's allowance of 20, and no plan bound + // (an unlimited plan or a self-hosted instance) admits the same send. + exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id,send_reserved_at,send_recipients)VALUES($1,'email',$2,'completed','',NOW()-INTERVAL '3 hours',$3)`, uuid.New(), f.recipient, recipients(5)) + plan := newTask("email", recipients(1)...) + exec(`INSERT INTO email_tasks(task_id,subject,body,in_reply_to)VALUES($1,'Re: Hello','Thanks',ARRAY[''])`, plan.TaskID) + var dayUsed int + if err := f.pool.QueryRow(ctx, `SELECT COALESCE(SUM(cardinality(t.send_recipients)),0) FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id + WHERE ea.organization_id=$1 AND t.task_type='email' AND t.send_reserved_at>NOW()-INTERVAL '24 hours' AND t.send_released_at IS NULL`, f.org).Scan(&dayUsed); err != nil { + t.Fatal(err) + } + plan.ManualPlanAllowance = dayUsed + if _, err := r.ReserveOutbound(ctx, plan); !errors.As(err, &limit) || limit.Window != models.ManualSendWindowPlan || limit.Limit != dayUsed { + t.Fatalf("send past the plan allowance = %v, want the plan bound", err) + } + plan.ManualPlanAllowance = dayUsed + 1 + if err := send(plan); err != nil { + t.Fatalf("send within the plan allowance refused: %v", err) + } + unbounded := newTask("email", recipients(1)...) + exec(`INSERT INTO email_tasks(task_id,subject,body,in_reply_to)VALUES($1,'Re: Hello','Thanks',ARRAY[''])`, unbounded.TaskID) + if err := send(unbounded); err != nil { + t.Fatalf("send with no plan bound refused: %v", err) + } + if got := SendAdmissionCode(&ManualSendLimitError{Window: models.ManualSendWindowDay, Limit: 1}); got != "MANUAL_SEND_LIMIT_REACHED" { t.Fatalf("code = %q", got) } diff --git a/internal/tasks/email_sender.go b/internal/tasks/email_sender.go index f3bd9b515..d37d66160 100644 --- a/internal/tasks/email_sender.go +++ b/internal/tasks/email_sender.go @@ -39,6 +39,9 @@ type EmailMessage struct { // bytes from object storage at send time. Refs travel inside the S3 body // blob, never the Avro Kafka event. Attachments []models.AttachmentRef + // ManualPlanAllowance is the workspace's plan-based unibox recipients in + // 24 hours; 0 means no plan bound. + ManualPlanAllowance int } // EmailSender interface for sending emails via workers @@ -187,7 +190,7 @@ func (s *emailSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessa recipients := append(append(append([]string{}, msg.To...), msg.CC...), msg.BCC...) nonce, err := s.admission.ReserveOutbound(ctx, repository.OutboundReservation{ TaskID: taskID, MailboxID: account.ID, OrganizationID: *account.OrganizationID, WorkerID: *workerID, - Provider: models.InboxProvider(account.Provider), Recipients: recipients, + Provider: models.InboxProvider(account.Provider), Recipients: recipients, ManualPlanAllowance: msg.ManualPlanAllowance, }) if err != nil { if msg.IsWarmup && msg.DispatchNonce != "" { diff --git a/internal/tasks/user_email_refusal_live_test.go b/internal/tasks/user_email_refusal_live_test.go index fdd845f76..fae6512ac 100644 --- a/internal/tasks/user_email_refusal_live_test.go +++ b/internal/tasks/user_email_refusal_live_test.go @@ -9,6 +9,8 @@ import ( "time" "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/feature" + "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" "github.com/warmbly/warmbly/internal/tasks/proto" @@ -101,3 +103,45 @@ func TestLiveUnrecordedUniboxRefusalIsRetried(t *testing.T) { t.Errorf("status = %q, want pending so the retry runs it again", status) } } + +// planGate answers the two gate questions a unibox send asks. +type planGate struct { + feature.FeatureGateService + daily int +} + +func (g planGate) CanUseUnibox(context.Context, uuid.UUID) (bool, *errx.Error) { return true, nil } + +func (g planGate) GetDailyEmailLimit(context.Context, uuid.UUID) (int, *errx.Error) { + return g.daily, nil +} + +// The workspace's plan sizes the allowance the send is reserved under: twice +// the plan's daily limit with a floor, and no bound on an unlimited plan or a +// self-hosted instance. +func TestLiveUniboxSendCarriesThePlanAllowance(t *testing.T) { + handle := liveCampaignDB(t) + f := newCampaignSendFixture(t, handle.Pool) + for _, tc := range []struct { + name string + daily int + want int + }{ + {"starter", 150, 300}, + {"free trial", 20, 100}, + {"unlimited or self-hosted", -1, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + sender := &recordingSender{} + svc := liveCampaignService(t, handle, sender) + svc.featureGate = planGate{daily: tc.daily} + taskID := queueUniboxReply(t, handle.Pool, f.mailbox, "", "") + if xerr := svc.HandleUserEmailTask(&proto.ProcessTask{TaskId: taskID.String()}); xerr != nil { + t.Fatalf("handle reply: %v", xerr) + } + if got := sender.message(t, 0).ManualPlanAllowance; got != tc.want { + t.Errorf("plan allowance = %d, want %d", got, tc.want) + } + }) + } +} diff --git a/internal/tasks/user_email_task.go b/internal/tasks/user_email_task.go index b229d7453..86709b0bf 100644 --- a/internal/tasks/user_email_task.go +++ b/internal/tasks/user_email_task.go @@ -152,6 +152,10 @@ func (s *tasksService) HandleUserEmailTask(task *proto.ProcessTask) *errx.Error } return s.refuseUserEmail(ctx, taskID, account, err) } + planAllowance, xerr := s.manualPlanAllowance(ctx, *account.OrganizationID) + if xerr != nil { + return xerr + } // STEP 5: Mark task as active (with advisory lock) if err := s.taskRepo.UpdateTaskStatusWithLock(ctx, taskID, "active"); err != nil { @@ -205,6 +209,8 @@ func (s *tasksService) HandleUserEmailTask(task *proto.ProcessTask) *errx.Error ThreadID: threadID, IsWarmup: false, Tracking: nil, + + ManualPlanAllowance: planAllowance, } if err := s.emailSender.Send(ctx, taskID, emailMsg, *account); err != nil { @@ -309,6 +315,20 @@ func userEmailRefusalReason(err error) string { } } +// manualPlanAllowance is the workspace's plan-based unibox allowance, read +// while the task is still pending so a failed read retries it. No gate (or a +// plan without a daily limit, or a self-hosted instance) means no plan bound. +func (s *tasksService) manualPlanAllowance(ctx context.Context, org uuid.UUID) (int, *errx.Error) { + if s.featureGate == nil { + return 0, nil + } + daily, xerr := s.featureGate.GetDailyEmailLimit(ctx, org) + if xerr != nil { + return 0, xerr + } + return models.ManualSendWorkspaceAllowance(daily), nil +} + func (s *tasksService) notifyUserEmailFailed(ctx context.Context, taskID uuid.UUID, account *models.Email, title, message string) { if s.streamingPublisher == nil { return