# Topic Inactivity GC — Design Spec

**Date:** 2026-03-29
**Status:** Draft
**Scope:** Auto-reap topics that have had no publish or consumer activity for a configurable duration.

## Problem

Topics are auto-created on first publish and live forever. When downstream projects abandon features, their topics remain in memory and on disk indefinitely — consuming resources with no consumers or producers. There is no built-in mechanism to reclaim these resources.

## Design Decisions

| Decision | Choice | Rationale |
|---|---|---|
| What counts as activity | Last publish OR last subscribe/fetch (whichever is newest) | Either side still caring means the topic is in use. Infrequent publishers (weekly/monthly) should not trigger reaping if consumers are still connecting. |
| Info queries count as activity | No | Observability queries are not consumption. |
| Config granularity | Namespace-level default + per-topic pattern overrides | Matches existing retention config model. Pattern overrides use existing NATS-style wildcards (`TopicPattern`) for dynamic topic naming conventions. |
| Default when unconfigured | No reaping (current behavior) | Opt-in only. Destructive operation must not surprise existing deployments on upgrade. |
| Active subscriber/inflight behavior | Topic is never reaped | Active subscribers or inflight messages mean the topic is definitionally active, regardless of timestamps. |
| Deletion model | Soft delete with 24h grace period (configurable) | Grace period gives operators time to notice and intervene before irreversible deletion. |
| Grace period state persistence | In-memory only | On restart, topics must re-qualify through the full inactivity window. This is intentionally conservative — the safe direction. |
| Implementation approach | Extend existing `RetentionTask::sweep()` | Reuses sweep infrastructure, config model, shutdown coordination. Segment retention and topic reaping are both "what happens to old/unused data." |

## Activity Tracking

### Engine Layer (`src/queue/topic.rs`)

New field on `Topic`:

```rust
last_activity_millis: AtomicU64  // epoch millis, Relaxed ordering
```

Updated on:
- `Topic::publish()` — producer wrote a message
- Engine-level `consume()` / `fetch()` paths — consumer read a message

On server startup during `engine.init()`, initialized from the newest message timestamp in the topic's buffer (recovered from storage replay via `Topic::from_replay()`). "Newest" means the last message in the replayed set after `max_depth` truncation — this is the correct value since truncated messages are not recoverable anyway.

`Topic::new()` initializes `last_activity_millis` to `now_millis()` unconditionally. For the `from_replay` path with non-empty messages, the value is then overwritten to the newest envelope's timestamp after construction. This means:
- Empty topic on disk (directory exists, zero messages after replay): gets `now_millis()` from `new()` — orphaned empty directories get a full inactivity window reprieve after every restart (intentionally conservative).
- Non-empty topic: gets the newest replayed message's timestamp.
- Freshly created topic (first publish): gets `now_millis()` from `new()`, immediately updated by `publish()`.

### Namespace Manager Layer (`src/namespace/state.rs`)

New field on `TopicDeliveryState`:

```rust
last_subscribe_millis: AtomicU64  // epoch millis, Relaxed ordering
```

Updated on:
- `NamespaceManager::subscribe()` / `subscribe_at()` / `subscribe_from()` — consumer connected
- Fetch operations routed through the manager

Initialized to 0 (no subscribers yet) on creation and on server restart.

### Effective Last Activity

```
effective_last_activity = max(topic.last_activity_millis, delivery_state.last_subscribe_millis)
```

Two separate timestamps respect the existing module boundary — `Topic` lives in the queue engine (no concept of subscriptions), `TopicDeliveryState` lives in the namespace manager. The retention task has access to both and computes the max.

## Configuration

### RetentionPolicy (`src/config.rs`)

New fields:

```rust
pub struct RetentionPolicy {
    // ... existing fields ...
    pub topic_inactive_secs: Option<u64>,          // None = no reaping (default)
    pub topic_reap_grace_period_secs: Option<u64>,  // None = 86400 (24h default)
}
```

`RetentionPolicy::is_active()` updated to include `topic_inactive_secs.is_some()`.

**Important:** The existing `RetentionTask::run()` exits early if no retention policies are active across all namespaces. With this change, a namespace that only has `topic_inactive_secs` set (no `max_age_secs`, `max_bytes`, or `max_segments`) will correctly keep the retention task alive because `is_active()` now includes the new field. This is a behavioral change to existing code — the early-exit guard at the top of `run()` depends on `is_active()`.

### Per-Topic Override

New struct:

```rust
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TopicInactiveOverride {
    pub pattern: String,    // NATS-style wildcard pattern, validated at config load
    pub inactive_secs: u64, // 0 = never reap topics matching this pattern
}
```

All override fields live inside `RetentionPolicy` to keep GC config in one struct:

```rust
pub struct RetentionPolicy {
    // ... existing fields ...
    pub topic_inactive_secs: Option<u64>,
    pub topic_reap_grace_period_secs: Option<u64>,
    #[serde(default)]
    pub topic_inactive_overrides: Vec<TopicInactiveOverride>,
}
```

### Override Resolution

Iterate `topic_inactive_overrides` in order. First matching pattern wins. If no pattern matches, fall back to `topic_inactive_secs`. `inactive_secs = 0` means "never reap."

Patterns are validated via `TopicPattern::parse()` at config load time (`RuntimeConfig::from_server_config`). Invalid patterns produce a config load error rather than silently failing during a sweep. Compiled `TopicPattern` values are cached in `RuntimeConfig` for the sweep to use directly — no re-parsing per topic per sweep.

### TOML Example

```toml
[namespaces.myapp.retention]
topic_inactive_secs = 604800          # 7 days default
topic_reap_grace_period_secs = 43200  # optional, defaults to 24h

[[namespaces.myapp.retention.topic_inactive_overrides]]
pattern = "system.>"
inactive_secs = 0                     # never reap system topics

[[namespaces.myapp.retention.topic_inactive_overrides]]
pattern = "tmp.*"
inactive_secs = 3600                  # 1 hour for temp topics
```

### Hot-Reload

Tier 1 — read per-sweep from `SharedConfig`. Changes take effect on the next sweep cycle with no coordination. Pattern compilation happens during config reload, not during sweep.

### RuntimeConfig

New fields carried via existing `namespace_retention` map through `RetentionPolicy`:
- `topic_inactive_secs`, `topic_reap_grace_period_secs`, and `topic_inactive_overrides` (with pre-compiled `TopicPattern` values)

## Reaping Logic

### Grace Period State

New field on `RetentionTask`:

```rust
grace_pending: HashMap<(NamespaceId, TopicName), Instant>  // when topic entered grace period
```

In-memory only. Empty on startup.

### Topic Discovery

The existing retention sweep discovers topics via `storage.list_topics()`, which only returns topics with persisted segment data on disk. Topics that exist only in memory (created by a recent publish, not yet flushed) are not eligible for inactivity reaping — a topic with no segments is by definition very recently created. This is the correct behavior and no change to the discovery source is needed.

### Sweep Flow (per topic, after segment retention)

```
1. Load namespace's topic_inactive_secs from config
   → None? Skip (no reaping configured)

2. Check topic_inactive_overrides — first matching pattern wins
   → Matched with inactive_secs = 0? Skip (exempt)
   → Matched? Use override's inactive_secs
   → No match? Use namespace default

3. Compute effective_last_activity = max(
     engine.topic_last_activity(ns, topic),
     manager.topic_last_subscribe(ns, topic)
   )

4. If now - effective_last_activity < threshold:
   → Topic is active
   → If in grace_pending: remove, log INFO "activity resumed"
   → Skip

5. Safety checks:
   → manager.has_active_subscribers(ns, topic)? Skip
   → manager.has_inflight_messages(ns, topic)? Skip

6. Not in grace_pending?
   → Insert with Instant::now()
   → Log WARN: "topic entered reap grace period"
   → Increment fafb_topic_reap_grace_pending gauge
   → Skip

7. In grace_pending and grace_elapsed >= grace_period?
   → REAP (see Reap Execution below)
```

### Reap Execution

```
1. Abort all delivery tasks for this topic (individual + group JoinHandles)
2. Remove TopicDeliveryState from namespace manager
   NOTE: Wildcard subscription entries in NamespaceState (subscription_trie,
   wildcard_subs, wildcard_group_subs) are NOT cleaned up. They survive topic
   reaping so that if the topic is re-created by a future publish,
   wildcard auto-subscribe will fire normally.
3. Flush/close the storage writer for this topic
   The engine may have unflushed dirty bytes. The storage layer's LRU fd cache
   may hold an open segment writer. This step ensures no open file handles or
   buffered data remain before deletion. Implemented via a new
   StorageBackend::close_topic_writer(ns, topic) method that flushes and evicts
   the topic's entry from the writer cache.
4. Remove Topic from engine's in-memory topic map, update total_bytes
   total_bytes (AtomicU64) is decremented via fetch_sub(topic.bytes_used() as u64, Relaxed)
   to match the existing increment pattern. Per-topic dirty_bytes is discarded
   (topic-local counter, not subtracted from any global).
5. Delete topic segment directory from storage (rm -rf the topic dir)
6. Decrement namespace topic_count
7. Remove from grace_pending
8. Log WARN: "topic reaped after {N}d inactivity, freed {bytes}B / {messages} messages"
9. Increment fafb_topic_reaped_total counter
10. Decrement fafb_topic_reap_grace_pending gauge
```

### Ordering

Segment retention runs first in the sweep, topic reaping runs second. Avoids wasted work cleaning up segments in a topic about to be deleted entirely.

## New Methods Required

| Layer | Method | Signature | Purpose |
|---|---|---|---|
| `QueueEngine` | `topic_last_activity` | `(ns, topic) -> Option<u64>` | Read `last_activity_millis` atomic |
| `QueueEngine` | `remove_topic` | `(ns, topic) -> Result<()>` | Remove from in-memory topic map, update `total_bytes` |
| `NamespaceManager` | `topic_last_subscribe` | `(ns, topic) -> Option<u64>` | Read `last_subscribe_millis` from delivery state |
| `NamespaceManager` | `has_active_subscribers` | `(ns, topic) -> bool` | Check individual + group subscriber count > 0 |
| `NamespaceManager` | `has_inflight_messages` | `(ns, topic) -> bool` | Check individual + group inflight maps non-empty |
| `NamespaceManager` | `teardown_topic` | `(ns, topic) -> Result<()>` | Abort delivery tasks, remove `TopicDeliveryState` |
| `StorageBackend` | `close_topic_writer` | `(ns, topic) -> Result<()>` | Flush and evict topic's writer from LRU fd cache |
| `StorageBackend` | `delete_topic_dir` | `(ns, topic) -> Result<()>` | Remove entire topic segment directory |

## Observability

### Metrics

| Metric | Type | Labels | Description |
|---|---|---|---|
| `fafb_topic_inactive_threshold_exceeded` | counter | `namespace` | Topics per sweep that exceeded the inactivity threshold (includes those entering grace, already in grace, and skipped due to active subscribers) |
| `fafb_topic_reap_grace_pending` | gauge | `namespace` | Topics currently in grace period |
| `fafb_topic_reaped_total` | counter | `namespace`, `topic` | Topics hard-deleted |
| `fafb_topic_reap_skipped_active` | counter | `namespace` | Reaping skipped due to active subscribers or inflight |

### Logging

- **WARN** — topic enters grace period (namespace, topic, last activity timestamp, threshold)
- **WARN** — topic hard-deleted (namespace, topic, days inactive)
- **INFO** — topic exits grace period due to resumed activity
- **DEBUG** — per-sweep summary (candidates evaluated, skipped, grace entries, reaped)

### Admin API

`GET /admin/topics/pending-reap` — returns topics currently in grace period:

```json
[
  {
    "namespace": "myapp",
    "topic": "abandoned-feature.events",
    "last_activity": "2026-03-15T10:30:00Z",
    "grace_entered": "2026-03-22T14:00:00Z",
    "grace_remaining_secs": 43200,
    "threshold_secs": 604800
  }
]
```

## Edge Cases

**DLQ topics:** Follow the same inactivity rules as any other topic. No linked lifecycle with parent — if parent is reaped and DLQ receives no further messages, DLQ becomes independently reap-eligible.

**Wildcard subscriptions:** Auto-subscribe timestamp from wildcard match counts as activity on the concrete topic. When wildcard subscriber disconnects and no other activity occurs, concrete topic becomes reap-eligible normally. Wildcard subscription entries in `NamespaceState` (`subscription_trie`, `wildcard_subs`, `wildcard_group_subs`) are NOT removed during topic reaping — only the per-topic `TopicDeliveryState` is torn down. This means if a reaped topic is re-created by a future publish, wildcard auto-subscribe fires normally and re-establishes delivery.

**Restart behavior:** Grace period state lost (in-memory). Topics must re-qualify through full inactivity window. `last_activity_millis` recovered from newest message timestamp during replay. `last_subscribe_millis` resets to 0. Strictly safer than persisting grace state.

**Race: publish during reap:** If a publish arrives for a topic in grace, the next sweep sees fresh `last_activity_millis` and removes it from `grace_pending`. If a publish arrives *during* reap execution, `get_or_create_topic()` re-creates the topic — historical data is lost, but it was already past the inactivity threshold.

**Empty topics:** `last_activity_millis` reflects last publish time even if all messages were deleted by segment retention. Reap removes the empty in-memory entry and segment directory.

**Config reload during grace:** Next sweep re-evaluates against new config. If topic no longer qualifies (threshold increased, override added), removed from `grace_pending`.

**Ordering in sweep:** Segment retention first, topic reaping second.

## Testing Strategy

### Unit Tests

- `Topic::publish()` updates `last_activity_millis`
- `last_activity_millis` initializes from newest message on construction
- `TopicDeliveryState::last_subscribe_millis` updates on subscribe
- `RetentionPolicy::is_active()` returns true when `topic_inactive_secs` is set
- Override pattern matching: first match wins, `inactive_secs = 0` exempts
- Config deserialization roundtrip with new fields
- `RuntimeConfig` diff detection for new fields
- `TopicInactiveOverride` pattern validation at config load time (invalid patterns rejected)
- `RetentionPolicy::is_active()` returns true with only `topic_inactive_secs` set (retention task stays alive)

### Integration Tests

- **Basic reap lifecycle:** publish → wait past threshold → grace entry → wait past grace → fully removed (memory + disk)
- **Publish resets clock:** publish, wait near threshold, publish again → not reaped
- **Subscribe/fetch resets clock:** publish, wait near threshold, subscribe/fetch → not reaped
- **Active subscriber blocks reap:** past threshold + connected subscriber → skip
- **Inflight blocks reap:** past threshold + unacked message → skip
- **Grace period recovery:** enter grace → new publish → removed from grace on next sweep
- **Override exemption:** `inactive_secs = 0` pattern → matching topic never reaped
- **Override precedence:** multiple patterns match → first-match wins
- **Config hot-reload:** topic in grace → reload with longer threshold → exits grace
- **DLQ independent lifecycle:** reap parent → DLQ becomes independently reap-eligible
- **DLQ cascade in same sweep:** both parent and DLQ inactive → both reaped in same sweep (ordering within sweep does not matter)
- **Retention task alive with only GC config:** namespace with `topic_inactive_secs` but no segment retention → retention task runs (does not early-exit)

### Deploy Integration Tests

- **Restart preserves safety:** topic near reap → restart → full inactivity window must re-elapse

Tests use short thresholds (1–2 seconds) and small grace periods for fast execution.
