feat: address review on segment-campaign linking: a campaign_lead_removals table (migration 000115, orgtransfer-registered) records hand-removed leads so the automatic segment sync never re-adds them while a manual add or the one-shot enrol clears the record, PUT /campaigns/:id/segments rejects an omitted segment_ids so {} cannot detach everything, rows.Err checks on the four new link queries so a truncated read cannot commit as success, write-path enrolment syncs detached from the request goroutine with a per-org in-flight dedupe, per-campaign 30s deadlines in the sweep instead of one shared budget, and an error-with-retry state in the linked-segments dialog

docs/content/docs/api/reference/campaigns.mdx

@@ -808,7 +808,7 @@ A `data` array of link objects (no pagination wrapper).
 
 `PUT /campaigns/:id/segments`
 
-Atomically replace the campaign's linked segments with the supplied set (up to 20). Every current member of a newly linked segment is enrolled as a lead immediately, and contacts who enter a linked segment later are enrolled automatically, within about 2 minutes. Enrolment is additive: a contact who leaves a segment keeps their lead row, and removing a segment from the set stops future enrolment without touching existing leads. An active campaign is woken to send to the new leads; a completed campaign is restarted through the full launch checks when a linked segment grows. A linked segment cannot be deleted (`DELETE /segments/:id` returns `409`) until it is removed here. **Scope** `WRITE_CAMPAIGNS` · **Org permission** `manage_campaigns`.
+Atomically replace the campaign's linked segments with the supplied set (up to 20). Every current member of a newly linked segment is enrolled as a lead immediately, and contacts who enter a linked segment later are enrolled automatically, within about 2 minutes. Enrolment is additive: a contact who leaves a segment keeps their lead row, and removing a segment from the set stops future enrolment without touching existing leads. A lead removed from the campaign by hand is never re-added automatically; a manual add (or the one-shot enrol below) clears that removal record. Omitting `segment_ids` returns `400`; send an explicit empty array to detach every segment. An active campaign is woken to send to the new leads; a completed campaign is restarted through the full launch checks when a linked segment grows. A linked segment cannot be deleted (`DELETE /segments/:id` returns `409`) until it is removed here. **Scope** `WRITE_CAMPAIGNS` · **Org permission** `manage_campaigns`.
 
 | Parameter | In | Type | Description |
 | --- | --- | --- | --- |

docs/content/docs/guides/campaigns.mdx

@@ -63,7 +63,7 @@ The **Leads** tab takes contacts four ways. **From contacts** opens a picker ove
 
 **Segments** on the same toolbar links [segments](/guides/segments/) to the campaign as a live audience, up to 20 per campaign. Linking enrols every current member immediately, and contacts who enter a linked segment later are enrolled on their own, within a couple of minutes. An active campaign wakes to send to them, a finished one restarts through the usual launch checks, and a paused one accumulates them for later. Enrolment is additive: a contact who leaves the segment keeps their lead row, and detaching a segment stops future enrolment without touching existing leads. See [linking a segment to a campaign](/guides/segments/#linking-a-segment-to-a-campaign).
 
-A lead can also be taken out again: the remove button on a lead's row, or **Remove from campaign** in the selection bar when several are ticked, drops the lead from this campaign without deleting the contact.
+A lead can also be taken out again: the remove button on a lead's row, or **Remove from campaign** in the selection bar when several are ticked, drops the lead from this campaign without deleting the contact. The removal sticks: a linked segment will not re-enrol that contact automatically, even while they still match it, until you add them back yourself.
 
 ### Lead statuses
 

docs/content/docs/guides/segments.mdx

@@ -59,7 +59,7 @@ Where **Add to campaign** copies today's members once, a linked segment is a liv
 
 Linking enrols every current member as a lead immediately. From then on, any contact who enters the segment, a new contact, an import, a condition edit, a manual pin-in, or simply drifting into a date or engagement condition, is enrolled automatically; a background check picks up drift about every 2 minutes.
 
-Enrolment is additive. A contact who later falls out of the segment keeps their lead row and their progress; the link only ever adds. Detaching a segment stops future enrolment but leaves the leads it already added in place.
+Enrolment is additive. A contact who later falls out of the segment keeps their lead row and their progress; the link only ever adds. Detaching a segment stops future enrolment but leaves the leads it already added in place. Removing a lead from the campaign by hand is respected too: automatic enrolment will not re-add that contact, even while they still match a linked segment, until you add them back yourself.
 
 The campaign reacts to growth by status:
 

internal/api/handler/segment.go

@@ -248,6 +248,12 @@ func (h *Handler) SetCampaignSegments(c *gin.Context) {
 		errx.Handle(c, errx.ErrInvalid)
 		return
 	}
+	// An omitted field must not read as "detach everything"; only an
+	// explicit [] does that.
+	if in.SegmentIDs == nil {
+		errx.Handle(c, errx.New(errx.BadRequest, "segment_ids is required; send [] to detach all segments"))
+		return
+	}
 	links, added, xerr := h.SegmentService.SetCampaignSegments(c.Request.Context(), orgID, campaignID, &in)
 	if xerr != nil {
 		errx.Handle(c, xerr)

internal/app/orgtransfer/spec.go

@@ -352,6 +352,11 @@ var Tables = []Table{
 		Name: "campaign_segments", Group: models.OrgDataGroupCampaigns,
 		Scope: `campaign_id IN ` + orgCampaigns,
 	},
+	{
+		Name: "campaign_lead_removals", Group: models.OrgDataGroupCampaigns,
+		Scope: `campaign_id IN ` + orgCampaigns,
+		Note:  "Must travel, or linked segments on the destination re-add every lead the user removed by hand.",
+	},
 	{
 		Name: "campaign_ab_assignments", Group: models.OrgDataGroupCampaigns,
 		Scope: `campaign_id IN ` + orgCampaigns,

internal/app/segment/service.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -68,6 +69,9 @@ type service struct {
 	fields  CustomFieldLister
 	waker   CampaignWaker
 	starter CampaignStarter
+	// orgSyncInFlight dedupes concurrent org-wide enrolment passes, keyed by
+	// org id.
+	orgSyncInFlight sync.Map
 }
 
 func NewService(repo repository.SegmentRepository, fields CustomFieldLister) Service {
@@ -388,25 +392,44 @@ func (s *service) syncLinkedCampaign(ctx context.Context, lc models.LinkedCampai
 
 // syncLinkedCampaignsForSegments re-enrols the campaigns linked to any of the
 // given segments. Nested references (a linked segment built on this one) are
-// not chased here; the periodic sweep covers them.
+// not chased here; the periodic sweep covers them. Detached from the caller's
+// request: the enrolment scans grow with the contact list, and this is
+// declared best effort with the sweep as the backstop.
 func (s *service) syncLinkedCampaignsForSegments(ctx context.Context, orgID uuid.UUID, segmentIDs []uuid.UUID) {
-	links, xerr := s.repo.LinkedCampaignsForSegments(ctx, orgID, segmentIDs)
-	if xerr != nil {
-		return
-	}
-	for _, lc := range links {
-		s.syncLinkedCampaign(ctx, lc)
-	}
+	bg := context.WithoutCancel(ctx)
+	go func() {
+		rctx, cancel := context.WithTimeout(bg, 2*time.Minute)
+		defer cancel()
+		links, xerr := s.repo.LinkedCampaignsForSegments(rctx, orgID, segmentIDs)
+		if xerr != nil {
+			return
+		}
+		for _, lc := range links {
+			s.syncLinkedCampaign(rctx, lc)
+		}
+	}()
 }
 
 func (s *service) SyncOrgLinkedCampaigns(ctx context.Context, orgID uuid.UUID) {
-	links, xerr := s.repo.LinkedCampaigns(ctx, &orgID)
-	if xerr != nil {
+	// One in-flight pass per org: every contact write calls this, and a burst
+	// of writes must not stack org-wide enrolment scans. A write landing while
+	// a pass runs is picked up by the sweep within its interval.
+	if _, running := s.orgSyncInFlight.LoadOrStore(orgID, struct{}{}); running {
 		return
 	}
-	for _, lc := range links {
-		s.syncLinkedCampaign(ctx, lc)
-	}
+	bg := context.WithoutCancel(ctx)
+	go func() {
+		defer s.orgSyncInFlight.Delete(orgID)
+		rctx, cancel := context.WithTimeout(bg, 2*time.Minute)
+		defer cancel()
+		links, xerr := s.repo.LinkedCampaigns(rctx, &orgID)
+		if xerr != nil {
+			return
+		}
+		for _, lc := range links {
+			s.syncLinkedCampaign(rctx, lc)
+		}
+	}()
 }
 
 func (s *service) StartCampaignSegmentSync(ctx context.Context, interval time.Duration) {
@@ -427,16 +450,23 @@ func (s *service) StartCampaignSegmentSync(ctx context.Context, interval time.Du
 }
 
 func (s *service) sweepLinkedCampaigns(ctx context.Context) {
-	rctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
-	defer cancel()
-	links, xerr := s.repo.LinkedCampaigns(rctx, nil)
+	scanCtx, scanCancel := context.WithTimeout(ctx, 30*time.Second)
+	links, xerr := s.repo.LinkedCampaigns(scanCtx, nil)
+	scanCancel()
 	if xerr != nil {
 		log.Warn().Str("error", xerr.Message).Msg("segment sync: sweep scan failed")
 		return
 	}
 	total := 0
+	// Each campaign gets its own deadline: one shared budget would starve the
+	// same tail campaigns every pass once the instance holds enough links.
 	for _, lc := range links {
-		total += s.syncLinkedCampaign(rctx, lc)
+		if ctx.Err() != nil {
+			return
+		}
+		cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+		total += s.syncLinkedCampaign(cctx, lc)
+		cancel()
 	}
 	if total > 0 {
 		log.Info().Int("added", total).Msg("segment sync: sweep enrolled new leads")

internal/infrastructure/db/migrations/000115_campaign_lead_removals.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS campaign_lead_removals;

internal/infrastructure/db/migrations/000115_campaign_lead_removals.up.sql

@@ -0,0 +1,10 @@
+-- A lead a user removed from a campaign by hand. Automatic segment enrolment
+-- must not re-add these pairs; an explicit manual add clears the record.
+CREATE TABLE campaign_lead_removals (
+    campaign_id uuid NOT NULL REFERENCES campaigns (id) ON DELETE CASCADE,
+    contact_id uuid NOT NULL REFERENCES contacts (id) ON DELETE CASCADE,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    PRIMARY KEY (campaign_id, contact_id)
+);
+
+CREATE INDEX idx_campaign_lead_removals_contact ON campaign_lead_removals (contact_id);

internal/repository/pg_contact.go

@@ -1828,14 +1828,23 @@ func (r *contactRepository) Update(ctx context.Context, userID, contactID string
 		toDelete := utils.Difference(currentCampaignIDs, wantIDs)
 
 		if len(toDelete) > 0 {
+			// Record the removal so linked-segment enrolment does not re-add
+			// the pair; a later manual add clears it again.
 			query = `
-				DELETE FROM campaign_leads cl
-				USING campaigns cam
-				WHERE cl.contact_id = $1
-				  AND cl.campaign_id = cam.id
-				  AND cam.id = ANY($2::uuid[])
-				  AND cam.organization_id = $3
-				RETURNING cl.campaign_id
+				WITH gone AS (
+					DELETE FROM campaign_leads cl
+					USING campaigns cam
+					WHERE cl.contact_id = $1
+					  AND cl.campaign_id = cam.id
+					  AND cam.id = ANY($2::uuid[])
+					  AND cam.organization_id = $3
+					RETURNING cl.contact_id, cl.campaign_id
+				), mark AS (
+					INSERT INTO campaign_lead_removals (campaign_id, contact_id)
+					SELECT campaign_id, contact_id FROM gone
+					ON CONFLICT (campaign_id, contact_id) DO UPDATE SET created_at = now()
+				)
+				SELECT campaign_id FROM gone
 			`
 			params := []any{contactID, toDelete, orgID}
 			rows, err := tx.Query(ctx, query, params...)
@@ -1853,6 +1862,14 @@ func (r *contactRepository) Update(ctx context.Context, userID, contactID string
 
 		if len(toInsert) > 0 {
 			query = `
+				WITH cleared AS (
+					DELETE FROM campaign_lead_removals r
+					USING campaigns cam
+					WHERE r.contact_id = $1
+					  AND r.campaign_id = cam.id
+					  AND cam.id = ANY($2::uuid[])
+					  AND cam.organization_id = $3
+				)
 				INSERT INTO campaign_leads (contact_id, campaign_id)
 				SELECT $1, cam.id
 				FROM campaigns cam
@@ -2109,7 +2126,11 @@ func (r *contactRepository) BulkUpdate(ctx context.Context, userID string, orgID
 		return nil
 	}
 	if len(data.RemoveCampaigns) > 0 {
-		if xerr := link(`DELETE FROM campaign_leads cl
+		// Only pairs that actually held a lead are recorded as removals, so
+		// the automatic segment enrolment never re-adds a hand-removed lead
+		// while contacts merely listed in the request stay eligible.
+		if xerr := link(`WITH gone AS (
+		         DELETE FROM campaign_leads cl
 		         USING contacts c, campaigns cam
 		         WHERE cl.contact_id = c.id
 		           AND cl.campaign_id = cam.id
@@ -2117,14 +2138,32 @@ func (r *contactRepository) BulkUpdate(ctx context.Context, userID string, orgID
 		           AND cam.organization_id = $1
 		           AND cl.contact_id = ANY($2)
 		           AND cl.campaign_id = ANY($3)
-		         RETURNING cl.contact_id, cl.campaign_id`,
+		         RETURNING cl.contact_id, cl.campaign_id
+		         ), mark AS (
+		         INSERT INTO campaign_lead_removals (campaign_id, contact_id)
+		         SELECT campaign_id, contact_id FROM gone
+		         ON CONFLICT (campaign_id, contact_id) DO UPDATE SET created_at = now()
+		         )
+		         SELECT contact_id, campaign_id FROM gone`,
 			models.ActivityCampaignRemoved, logCampaignLinks, orgID, data.Contacts, data.RemoveCampaigns); xerr != nil {
 			return nil, xerr
 		}
 	}
 
 	if len(data.AddCampaigns) > 0 {
-		if xerr := link(`INSERT INTO campaign_leads (contact_id, campaign_id)
+		// A manual add clears the removal record: the user changed their mind,
+		// so linked segments may manage this pair again.
+		if xerr := link(`WITH cleared AS (
+		         DELETE FROM campaign_lead_removals r
+		         USING contacts c, campaigns cam
+		         WHERE r.contact_id = c.id
+		           AND r.campaign_id = cam.id
+		           AND c.organization_id = $1
+		           AND cam.organization_id = $1
+		           AND r.contact_id = ANY($2)
+		           AND r.campaign_id = ANY($3::uuid[])
+		         )
+		         INSERT INTO campaign_leads (contact_id, campaign_id)
 		         SELECT c.id, cam.id
 		         FROM contacts c
 		         CROSS JOIN campaigns cam

internal/repository/pg_segment.go

@@ -299,7 +299,7 @@ func (r *segmentRepository) AddToCampaign(ctx context.Context, orgID uuid.UUID,
 		return nil, errx.InternalError()
 	}
 
-	links, err := insertSegmentLeads(ctx, tx, orgID, actorID(actor), clause, args, campaignID)
+	links, err := insertSegmentLeads(ctx, tx, orgID, actorID(actor), clause, args, campaignID, false)
 	if err != nil {
 		db.CaptureError(err, "segment enrol", nil, "query")
 		return nil, errx.InternalError()
@@ -314,12 +314,26 @@ func (r *segmentRepository) AddToCampaign(ctx context.Context, orgID uuid.UUID,
 // insertSegmentLeads enrols every contact matching the precompiled segment
 // clause as a lead, logging a campaign_added activity for each row that was
 // actually new. The campaign is bound after the clause's own parameters.
-func insertSegmentLeads(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, actor *uuid.UUID, clause string, args []any, campaignID uuid.UUID) ([]contactLink, error) {
+//
+// respectRemovals decides what a manual "remove from campaign" means here:
+// the automatic sync honours the removal record and skips the pair, while an
+// explicit enrol (the one-shot add-to-campaign) clears it and re-adds.
+func insertSegmentLeads(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, actor *uuid.UUID, clause string, args []any, campaignID uuid.UUID, respectRemovals bool) ([]contactLink, error) {
 	args = append(args, campaignID)
+	guard := ""
+	if respectRemovals {
+		guard = fmt.Sprintf(` AND NOT EXISTS (SELECT 1 FROM campaign_lead_removals r WHERE r.campaign_id = $%d AND r.contact_id = c.id)`, len(args))
+	} else {
+		clearQ := fmt.Sprintf(`DELETE FROM campaign_lead_removals r
+			WHERE r.campaign_id = $%d AND r.contact_id IN (SELECT c.id FROM contacts c WHERE c.organization_id = $1 AND (%s))`, len(args), clause)
+		if _, err := tx.Exec(ctx, clearQ, args...); err != nil {
+			return nil, err
+		}
+	}
 	insertQ := fmt.Sprintf(`INSERT INTO campaign_leads (contact_id, campaign_id)
-		SELECT c.id, $%d::uuid FROM contacts c WHERE c.organization_id = $1 AND (%s)
+		SELECT c.id, $%d::uuid FROM contacts c WHERE c.organization_id = $1 AND (%s)%s
 		ON CONFLICT DO NOTHING
-		RETURNING contact_id, campaign_id`, len(args), clause)
+		RETURNING contact_id, campaign_id`, len(args), clause, guard)
 	rows, err := tx.Query(ctx, insertQ, args...)
 	if err != nil {
 		return nil, err
@@ -377,6 +391,12 @@ func (r *segmentRepository) ListForCampaign(ctx context.Context, orgID, campaign
 		matches = append(matches, models.SegmentMatch(match))
 		conds = append(conds, cs)
 	}
+	// A mid-stream read failure ends Next() early with no scan error; without
+	// this the Leads tab would render a truncated link list as the truth.
+	if err := rows.Err(); err != nil {
+		db.CaptureError(err, "campaign segments list", nil, "rows")
+		return nil, errx.InternalError()
+	}
 	for i := range out {
 		n, xerr := r.Count(ctx, orgID, &out[i].SegmentID, matches[i], conds[i])
 		if xerr != nil {
@@ -461,6 +481,12 @@ func (r *segmentRepository) SyncCampaignSegments(ctx context.Context, orgID, cam
 		segmentIDs = append(segmentIDs, id)
 	}
 	rows.Close()
+	// A truncated id list here would enrol part of the audience and still
+	// commit as a success, so a read failure has to abort the sync.
+	if err := rows.Err(); err != nil {
+		db.CaptureError(err, "campaign segments sync", nil, "rows")
+		return 0, errx.InternalError()
+	}
 
 	total := 0
 	for _, segID := range segmentIDs {
@@ -474,7 +500,7 @@ func (r *segmentRepository) SyncCampaignSegments(ctx context.Context, orgID, cam
 		if clause == "FALSE" {
 			continue
 		}
-		links, lerr := insertSegmentLeads(ctx, tx, orgID, nil, clause, args, campaignID)
+		links, lerr := insertSegmentLeads(ctx, tx, orgID, nil, clause, args, campaignID, true)
 		if lerr != nil {
 			db.CaptureError(lerr, "segment enrol", nil, "query")
 			return 0, errx.InternalError()
@@ -505,6 +531,10 @@ func scanLinkedCampaigns(rows pgx.Rows) ([]models.LinkedCampaign, *errx.Error) {
 		}
 		out = append(out, l)
 	}
+	if err := rows.Err(); err != nil {
+		db.CaptureError(err, "linked campaigns", nil, "rows")
+		return nil, errx.InternalError()
+	}
 	return out, nil
 }
 
@@ -555,6 +585,10 @@ func (r *segmentRepository) CampaignsUsingSegment(ctx context.Context, orgID, se
 		}
 		names = append(names, n)
 	}
+	if err := rows.Err(); err != nil {
+		db.CaptureError(err, "campaigns using segment", nil, "rows")
+		return nil, errx.InternalError()
+	}
 	return names, nil
 }
 

internal/repository/segment_live_test.go

@@ -286,6 +286,38 @@ func TestLiveSegmentCampaignLinks(t *testing.T) {
 		t.Fatalf("post-include sync = %d, want 1", added)
 	}
 
+	// A hand-removed lead stays removed: the sync skips the pair until a
+	// manual add clears the record, and the explicit one-shot enrol overrides
+	// and clears it too.
+	handle, _ := liveContactDB(t)
+	contacts := NewContactRepostory(handle)
+	removeBob := &models.BulkEditContactsData{Contacts: []string{f.bob.String()}, RemoveCampaigns: []string{f.other.String()}}
+	if _, xerr := contacts.BulkUpdate(ctx, f.owner.String(), f.org, removeBob); xerr != nil {
+		t.Fatalf("remove: %v", xerr)
+	}
+	if added, _ = repo.SyncCampaignSegments(ctx, f.org, f.other); added != 0 {
+		t.Fatalf("sync after manual removal = %d, want 0", added)
+	}
+	if _, xerr := contacts.BulkUpdate(ctx, f.owner.String(), f.org, &models.BulkEditContactsData{
+		Contacts: []string{f.bob.String()}, AddCampaigns: []string{f.other.String()},
+	}); xerr != nil {
+		t.Fatalf("re-add: %v", xerr)
+	}
+	var removals int
+	if err := handle.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_lead_removals WHERE campaign_id = $1`, f.other).Scan(&removals); err != nil || removals != 0 {
+		t.Fatalf("removals after manual re-add = %d, %v", removals, err)
+	}
+	if _, xerr := contacts.BulkUpdate(ctx, f.owner.String(), f.org, removeBob); xerr != nil {
+		t.Fatalf("remove again: %v", xerr)
+	}
+	out, xerr := repo.AddToCampaign(ctx, f.org, f.owner.String(), acme.ID, f.other)
+	if xerr != nil || out.Added != 1 {
+		t.Fatalf("one-shot after removal = %+v, %v", out, xerr)
+	}
+	if err := handle.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_lead_removals WHERE campaign_id = $1`, f.other).Scan(&removals); err != nil || removals != 0 {
+		t.Fatalf("removals after one-shot = %d, %v", removals, err)
+	}
+
 	// The sweep and the targeted lookup both see the linked campaign, and the
 	// delete guard reports it by name.
 	linked, xerr := repo.LinkedCampaigns(ctx, &f.org)

web/src/components/app/segments/CampaignSegmentsDialog.tsx

@@ -108,6 +108,12 @@ export default function CampaignSegmentsDialog({
     }
 
     const loading = segments.isPending || (linked.isPending && !seeded);
+    // A failed load must not strand the dialog as "loaded but Save disabled".
+    const loadError = segments.isError || (linked.isError && !seeded);
+    const retryLoad = () => {
+        if (segments.isError) void segments.refetch();
+        if (linked.isError) void linked.refetch();
+    };
 
     return (
         <AnimatePresence>
@@ -168,6 +174,18 @@ export default function CampaignSegmentsDialog({
                                         <div key={i} className="h-9 rounded-md bg-slate-100 animate-pulse" />
                                     ))}
                                 </div>
+                            ) : loadError ? (
+                                <div className="px-5 py-10 text-center">
+                                    <p className="text-[12.5px] text-slate-900 font-medium">Couldn't load segments</p>
+                                    <p className="text-[11.5px] text-slate-400 mt-0.5">Check your connection and try again.</p>
+                                    <button
+                                        type="button"
+                                        onClick={retryLoad}
+                                        className="mt-3 h-7 px-2.5 rounded-md border border-slate-200 hover:border-slate-300 text-[12px] text-slate-700 hover:text-slate-900 transition-colors"
+                                    >
+                                        Retry
+                                    </button>
+                                </div>
                             ) : list.length === 0 ? (
                                 <div className="px-5 py-10 text-center">
                                     <p className="text-[12.5px] text-slate-900 font-medium">{query ? "No segments match" : "No segments yet"}</p>