From c9318d7bb026bf8032018fab5cff66f99e32c6ab Mon Sep 17 00:00:00 2001 From: prosolis <5590409+prosolis@users.noreply.github.com> Date: Fri, 22 May 2026 18:51:33 -0700 Subject: [PATCH] Hard daily cap, no-flood shutdown, ctx-aware poller, double-image fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 1 + internal/config/config.go | 3 ++ internal/ingestion/poller.go | 17 +++++++--- internal/matrix/client.go | 21 +++++++----- internal/poster/queue.go | 66 +++++++++++++++++++++--------------- internal/storage/queries.go | 11 ++++++ main.go | 2 +- 7 files changed, 80 insertions(+), 41 deletions(-) diff --git a/config.example.yaml b/config.example.yaml index 722b1d5..7059583 100644 --- a/config.example.yaml +++ b/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" diff --git a/internal/config/config.go b/internal/config/config.go index 4ecd422..cb17ba4 100644 --- a/internal/config/config.go +++ b/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 { diff --git a/internal/ingestion/poller.go b/internal/ingestion/poller.go index c2453c0..846b632 100644 --- a/internal/ingestion/poller.go +++ b/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 { diff --git a/internal/matrix/client.go b/internal/matrix/client.go index cbf0943..38068fe 100644 --- a/internal/matrix/client.go +++ b/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. diff --git a/internal/poster/queue.go b/internal/poster/queue.go index 1290fe3..0f9770e 100644 --- a/internal/poster/queue.go +++ b/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) - } - 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) + 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.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]...) diff --git a/internal/storage/queries.go b/internal/storage/queries.go index 9716804..d59de59 100644 --- a/internal/storage/queries.go +++ b/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 diff --git a/main.go b/main.go index 1a5a12c..ee28e4a 100644 --- a/main.go +++ b/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)