fix(traces): review fixes for trace_iatas

Refresh last_heard on duplicate observations too, capped at hourly, so
steady traffic can't age a trace out of the filter (dedup key has no
heard_at). Drop the unused last_heard index column so upserts stay HOT,
and share one retention cutoff across the cleanup deletes.

db/migrations/012_trace_iatas.sql

@@ -8,7 +8,7 @@ CREATE TABLE trace_iatas (
   PRIMARY KEY (trace_tag, iata)
 );
 
-CREATE INDEX idx_trace_iatas_iata ON trace_iatas(iata, last_heard DESC);
+CREATE INDEX idx_trace_iatas_iata ON trace_iatas(iata);
 
 -- Seed from retained packets; parallelism off so the join spills to disk, not /dev/shm.
 SET max_parallel_workers_per_gather = 0;

db/queries/queries.sql

@@ -684,10 +684,12 @@ ON CONFLICT (channel_hash, iata) DO UPDATE SET
 WHERE EXCLUDED.last_heard > channel_iatas.last_heard + INTERVAL '1 hour';
 
 -- name: UpsertTraceIATA :exec
+-- Refreshes at most hourly so repeat hears don't churn the row.
 INSERT INTO trace_iatas (trace_tag, iata, last_heard)
 VALUES ($1, $2, $3)
 ON CONFLICT (trace_tag, iata) DO UPDATE SET
-  last_heard = GREATEST(trace_iatas.last_heard, EXCLUDED.last_heard);
+  last_heard = EXCLUDED.last_heard
+WHERE EXCLUDED.last_heard > trace_iatas.last_heard + INTERVAL '1 hour';
 
 -- name: ListChannels :many
 -- Channels ordered by last seen, optionally filtered by hash and/or IATAs

db/sqlc/querier.go

@@ -236,6 +236,7 @@ type Querier interface {
 	UpsertPacket(ctx context.Context, arg UpsertPacketParams) (UpsertPacketRow, error)
 	UpsertRegion(ctx context.Context, arg UpsertRegionParams) (int32, error)
 	UpsertRegionIATA(ctx context.Context, arg UpsertRegionIATAParams) error
+	// Refreshes at most hourly so repeat hears don't churn the row.
 	UpsertTraceIATA(ctx context.Context, arg UpsertTraceIATAParams) error
 	// ============================================================
 	// TRANSPORT CODES

db/sqlc/queries.sql.go

@@ -4072,7 +4072,8 @@ const upsertTraceIATA = `-- name: UpsertTraceIATA :exec
 INSERT INTO trace_iatas (trace_tag, iata, last_heard)
 VALUES ($1, $2, $3)
 ON CONFLICT (trace_tag, iata) DO UPDATE SET
-  last_heard = GREATEST(trace_iatas.last_heard, EXCLUDED.last_heard)
+  last_heard = EXCLUDED.last_heard
+WHERE EXCLUDED.last_heard > trace_iatas.last_heard + INTERVAL '1 hour'
 `
 
 type UpsertTraceIATAParams struct {
@@ -4081,6 +4082,7 @@ type UpsertTraceIATAParams struct {
 	LastHeard pgtype.Timestamptz `json:"last_heard"`
 }
 
+// Refreshes at most hourly so repeat hears don't churn the row.
 func (q *Queries) UpsertTraceIATA(ctx context.Context, arg UpsertTraceIATAParams) error {
 	_, err := q.db.Exec(ctx, upsertTraceIATA, arg.TraceTag, arg.Iata, arg.LastHeard)
 	return err

internal/background/tasks.go

@@ -41,13 +41,16 @@ func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval
 			if err := store.DeleteOldTelemetry(ctx, time.Now().Add(-telemetryRetention)); err != nil {
 				return err
 			}
-			if err := store.DeleteOldPackets(ctx, time.Now().Add(-packetRetention)); err != nil {
+			// One cutoff for all three so the IATA tables stay in step
+			// with the packets they mirror.
+			cutoff := time.Now().Add(-packetRetention)
+			if err := store.DeleteOldPackets(ctx, cutoff); err != nil {
 				return err
 			}
-			if err := store.DeleteOldChannelIATAs(ctx, time.Now().Add(-packetRetention)); err != nil {
+			if err := store.DeleteOldChannelIATAs(ctx, cutoff); err != nil {
 				return err
 			}
-			if err := store.DeleteOldTraceIATAs(ctx, time.Now().Add(-packetRetention)); err != nil {
+			if err := store.DeleteOldTraceIATAs(ctx, cutoff); err != nil {
 				return err
 			}
 			return nil

internal/ingest/packet.go

@@ -790,7 +790,8 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		}
 	}
 
-	if traceTag != nil && inserted {
+	// Runs on duplicate observations too; the upsert only writes when the row is >1h stale.
+	if traceTag != nil {
 		if err := w.db.UpsertTraceIATA(ctx, traceTag, iata, heardAt); err != nil {
 			log.Printf("ingest[%s]: db: upsert trace IATA failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
 		}