Hard daily cap, no-flood shutdown, ctx-aware poller, double-image fix

Four related fixes after Pete flooded a channel and ignored Ctrl-C:

1. Global daily cap (posting.daily_cap_total, default 5): hard ceiling on
   posts across ALL channels in a rolling 24h window. Checked before the
   per-channel min-interval and burst-cap.

2. Shutdown no longer flushes the queue. Previous drainAll posted every
   remaining item with rate limits disabled — which was literally the
   flood. Replaced with dropOnShutdown that clears queues and logs
   the count.

3. Poller respects ctx mid-loop. pollOnceWithErr now takes ctx and bails
   between items, so Ctrl-C doesn't have to wait for ~30s of network
   per pending story before shutdown can complete.

4. Double-image fix. PostStory now reports imageSent; the queue clears
   ImageURL before retry so a text-send failure after a successful
   image upload doesn't re-post the image.

config.example.yaml

@@ -20,6 +20,7 @@ posting:
   min_interval_seconds: 300
   burst_cap_count: 3
   burst_cap_window_seconds: 1800
+  daily_cap_total: 5  # hard global cap across ALL channels (rolling 24h); 0 disables
 
 storage:
   db_path: "./data/pete.db"

internal/config/config.go

@@ -42,6 +42,9 @@ type PostingConfig struct {
 	BurstCapCount         int `yaml:"burst_cap_count"`
 	BurstCapWindowSeconds int `yaml:"burst_cap_window_seconds"`
 	DedupCooldownHours    int `yaml:"dedup_cooldown_hours"`
+	// DailyCapTotal is the hard global cap on posts across ALL channels in a
+	// rolling 24-hour window. 0 disables the cap.
+	DailyCapTotal int `yaml:"daily_cap_total"`
 }
 
 type StorageConfig struct {

internal/ingestion/poller.go

@@ -56,7 +56,7 @@ func (p *Poller) pollSource(ctx context.Context, src config.SourceConfig) {
 	slog.Info("starting poller", "source", src.Name, "interval", interval)
 
 	// Poll immediately on start, then on ticker
-	p.pollOnce(src)
+	p.pollOnce(ctx, src)
 
 	consecutiveFailures := 0
 	const adminWarningThreshold = 5
@@ -67,7 +67,7 @@ func (p *Poller) pollSource(ctx context.Context, src config.SourceConfig) {
 			slog.Info("poller stopping", "source", src.Name)
 			return
 		case <-ticker.C:
-			if err := p.pollOnceWithErr(src); err != nil {
+			if err := p.pollOnceWithErr(ctx, src); err != nil {
 				consecutiveFailures++
 				slog.Error("feed poll failed",
 					"source", src.Name,
@@ -88,13 +88,13 @@ func (p *Poller) pollSource(ctx context.Context, src config.SourceConfig) {
 	}
 }
 
-func (p *Poller) pollOnce(src config.SourceConfig) {
-	if err := p.pollOnceWithErr(src); err != nil {
+func (p *Poller) pollOnce(ctx context.Context, src config.SourceConfig) {
+	if err := p.pollOnceWithErr(ctx, src); err != nil {
 		slog.Error("feed poll failed", "source", src.Name, "err", err)
 	}
 }
 
-func (p *Poller) pollOnceWithErr(src config.SourceConfig) error {
+func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) error {
 	items, err := FetchFeed(src.FeedURL)
 	if err != nil {
 		return err
@@ -102,6 +102,10 @@ func (p *Poller) pollOnceWithErr(src config.SourceConfig) error {
 
 	newCount := 0
 	for i := range items {
+		if ctx.Err() != nil {
+			slog.Info("poller: cancellation observed mid-loop", "source", src.Name, "processed", newCount)
+			return nil
+		}
 		if storage.IsGUIDSeen(items[i].GUID) {
 			continue
 		}
@@ -193,6 +197,9 @@ func (p *Poller) pollOnceWithErr(src config.SourceConfig) error {
 		return nil
 	}
 	for _, s := range unclassified {
+		if ctx.Err() != nil {
+			return nil
+		}
 		// Skip stories we just ingested (they're already being processed above)
 		alreadyProcessed := false
 		for _, item := range items {

internal/matrix/client.go

@@ -323,12 +323,15 @@ func (c *Client) PostThreadedReply(channel string, rootEventID id.EventID, plain
 	return err
 }
 
-// PostStory sends a story to a channel. Returns the text event ID for reaction tracking.
-// If the story has a validated image, it's uploaded and sent as a separate m.image event first.
-func (c *Client) PostStory(channel string, story *PostableStory) (id.EventID, error) {
+// PostStory sends a story to a channel. Returns the text event ID for
+// reaction tracking and a flag indicating whether the m.image event was
+// successfully sent. Callers retrying after a failed text send MUST clear
+// the ImageURL on the next attempt if imageSent is true, otherwise the
+// image will be posted twice.
+func (c *Client) PostStory(channel string, story *PostableStory) (eventID id.EventID, imageSent bool, err error) {
 	roomID, ok := c.channels[channel]
 	if !ok {
-		return "", fmt.Errorf("unknown channel: %s", channel)
+		return "", false, fmt.Errorf("unknown channel: %s", channel)
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
@@ -336,8 +339,10 @@ func (c *Client) PostStory(channel string, story *PostableStory) (id.EventID, er
 
 	// Send image first if present
 	if story.ImageURL != "" {
-		if err := c.sendImage(ctx, roomID, story.ImageURL); err != nil {
-			slog.Warn("failed to send image, posting text only", "url", story.ImageURL, "err", err)
+		if ierr := c.sendImage(ctx, roomID, story.ImageURL); ierr != nil {
+			slog.Warn("failed to send image, posting text only", "url", story.ImageURL, "err", ierr)
+		} else {
+			imageSent = true
 		}
 	}
 
@@ -352,10 +357,10 @@ func (c *Client) PostStory(channel string, story *PostableStory) (id.EventID, er
 
 	resp, err := c.mx.SendMessageEvent(ctx, roomID, event.EventMessage, content)
 	if err != nil {
-		return "", fmt.Errorf("send message: %w", err)
+		return "", imageSent, fmt.Errorf("send message: %w", err)
 	}
 
-	return resp.EventID, nil
+	return resp.EventID, imageSent, nil
 }
 
 // sendImage downloads an image, uploads it to Matrix, and sends it as m.image.

internal/poster/queue.go

@@ -66,7 +66,9 @@ func (q *Queue) Enqueue(item QueueItem) {
 	)
 }
 
-// Start runs the queue drain ticker. Blocks until ctx is cancelled, then drains remaining items.
+// Start runs the queue drain ticker. Blocks until ctx is cancelled, then
+// drops any remaining queued items (no flush-posting on shutdown — we'd
+// rather lose them than dump a flood into the channel).
 func (q *Queue) Start(ctx context.Context) {
 	defer close(q.done)
 	ticker := time.NewTicker(10 * time.Second)
@@ -75,8 +77,7 @@ func (q *Queue) Start(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			// Graceful drain: attempt to post remaining queued items
-			q.drainAll()
+			q.dropOnShutdown()
 			return
 		case <-ticker.C:
 			q.drain()
@@ -102,29 +103,23 @@ func (q *Queue) drain() {
 	}
 }
 
-// drainAll posts all remaining items ignoring rate limits (shutdown path).
-func (q *Queue) drainAll() {
+// dropOnShutdown clears any pending queues, logging the count. We deliberately
+// do NOT post these — the daily cap exists for a reason, and a flush on Ctrl-C
+// would dump everything pending into the rooms at once.
+func (q *Queue) dropOnShutdown() {
 	q.mu.Lock()
-	channels := make([]string, 0, len(q.queues))
-	for ch := range q.queues {
-		channels = append(channels, ch)
+	defer q.mu.Unlock()
+	total := 0
+	for ch, items := range q.queues {
+		if len(items) > 0 {
+			slog.Info("dropping queued items on shutdown",
+				"channel", ch, "count", len(items))
+			total += len(items)
 		}
-	q.mu.Unlock()
-
-	for _, ch := range channels {
-		for {
-			q.mu.Lock()
-			items := q.queues[ch]
-			if len(items) == 0 {
-				q.mu.Unlock()
-				break
-			}
-			item := items[0]
-			q.queues[ch] = items[1:]
-			q.mu.Unlock()
-
-			q.postItem(item)
+		q.queues[ch] = nil
 	}
+	if total > 0 {
+		slog.Info("shutdown: queue drop complete", "total_dropped", total)
 	}
 }
 
@@ -141,13 +136,25 @@ func (q *Queue) drainChannel(channel string) {
 	minInterval := int64(q.config.MinIntervalSeconds)
 	burstWindow := int64(q.config.BurstCapWindowSeconds)
 
-	// Check minimum interval since last post
+	// Global daily cap: hard ceiling across ALL channels (rolling 24h).
+	// Checked first so it short-circuits per-channel logic.
+	if q.config.DailyCapTotal > 0 {
+		dayStart := now - 24*3600
+		todays := storage.CountAllPostsInWindow(dayStart)
+		if todays >= q.config.DailyCapTotal {
+			slog.Debug("global daily cap reached, holding queue",
+				"channel", channel, "posts_24h", todays, "cap", q.config.DailyCapTotal)
+			return
+		}
+	}
+
+	// Check minimum interval since last post (per channel)
 	lastPost := storage.GetLastPostTime(channel)
 	if now-lastPost < minInterval {
 		return
 	}
 
-	// Check burst cap
+	// Check burst cap (per channel)
 	windowStart := now - burstWindow
 	postsInWindow := storage.CountPostsInWindow(channel, windowStart)
 	if postsInWindow >= q.config.BurstCapCount {
@@ -194,7 +201,7 @@ func (q *Queue) postItem(item QueueItem) {
 		Platforms:  item.Platforms,
 	}
 
-	eventID, err := q.mx.PostStory(item.Channel, story)
+	eventID, imageSent, err := q.mx.PostStory(item.Channel, story)
 	if err != nil {
 		item.retries++
 		if item.retries >= maxRetries {
@@ -211,6 +218,11 @@ func (q *Queue) postItem(item QueueItem) {
 			"attempt", item.retries,
 			"err", err,
 		)
+		// If the image already went up before the text failed, clear it on the
+		// retry so we don't send the m.image event a second time.
+		if imageSent {
+			item.ImageURL = ""
+		}
 		// Re-queue at front for retry
 		q.mu.Lock()
 		q.queues[item.Channel] = append([]QueueItem{item}, q.queues[item.Channel]...)

internal/storage/queries.go

@@ -182,6 +182,17 @@ func GetLastPostTime(channel string) int64 {
 	return t
 }
 
+// CountAllPostsInWindow counts posts across ALL channels within a time window.
+// Used for the global daily cap.
+func CountAllPostsInWindow(windowStart int64) int {
+	var count int
+	if err := Get().QueryRow(`SELECT COUNT(*) FROM post_log WHERE posted_at >= ?`, windowStart).Scan(&count); err != nil {
+		slog.Error("CountAllPostsInWindow query failed", "err", err)
+		return 1<<31 - 1 // fail closed: huge number prevents posting
+	}
+	return count
+}
+
 // CountPostsInWindow counts posts to a channel within a time window.
 func CountPostsInWindow(channel string, windowStart int64) int {
 	var count int

main.go

@@ -272,7 +272,7 @@ func runTest(cfg *config.Config, sourceName string) {
 		Channel:    channel,
 	}
 
-	eventID, err := mx.PostStory(channel, story)
+	eventID, _, err := mx.PostStory(channel, story)
 	if err != nil {
 		slog.Error("test: post failed", "err", err)
 		os.Exit(1)