diff --git a/internal/store/ack.go b/internal/store/ack.go index 132499f..e910174 100644 --- a/internal/store/ack.go +++ b/internal/store/ack.go @@ -2,6 +2,7 @@ package store import ( "context" + "database/sql" "fmt" "time" ) @@ -38,16 +39,21 @@ func (s *Store) MarkSent(ctx context.Context, key string, ts time.Time) error { // send goes through RecordNudge, not MarkSent, so the first MarkSent comes on // the repeat path — LastSent may legitimately be zero until then). func (s *Store) LastSent(ctx context.Context, key string) (time.Time, error) { - var millis int64 + // MAX over an empty set is one row holding NULL, not zero rows, so this + // scans into a NullInt64 — the same trap OldestPendingTelegram spells out. + // A bare int64 turned the ordinary "nothing sent yet" case into a scan + // error, and RepeatUnacked aborts its whole sweep on one, so the first + // repeat could never go out for any rule. + var millis sql.NullInt64 err := s.db.QueryRowContext(ctx, `SELECT MAX(sent_at) FROM ack_sends WHERE rule = ?`, key).Scan(&millis) if err != nil { return time.Time{}, fmt.Errorf("last sent %s: %w", key, err) } - if millis == 0 { + if !millis.Valid { return time.Time{}, nil } - return time.UnixMilli(millis).UTC(), nil + return time.UnixMilli(millis.Int64).UTC(), nil } // MarkAcked marks ALL pending telegram nudges for the rule as "acted" — diff --git a/internal/store/ack_test.go b/internal/store/ack_test.go new file mode 100644 index 0000000..8c215e9 --- /dev/null +++ b/internal/store/ack_test.go @@ -0,0 +1,60 @@ +package store + +import ( + "context" + "testing" + "time" +) + +// TestLastSentOnEmptyTableIsZero pins the aggregate-over-nothing trap that +// nudges.go's OldestPendingTelegram already documents: MAX over an empty set is +// one row holding NULL, not zero rows. Scanning that into a bare int64 is an +// error, and the doc on LastSent promises a zero time instead. +// +// It is not a cosmetic promise. ack_sends is written only by MarkSent, and +// MarkSent is called only after a repeat has already gone out, so the first +// repeat for every rule reads an empty table. delivery.Dispatcher.RepeatUnacked +// aborts the whole sweep on that error, which means the repeat-til-ack loop can +// never take its first step for any rule. +func TestLastSentOnEmptyTableIsZero(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + last, err := s.LastSent(ctx, "service_down") + if err != nil { + t.Fatalf("last sent on an empty table must not error: %v", err) + } + if !last.IsZero() { + t.Fatalf("want the zero time before anything was sent, got %v", last) + } +} + +// TestLastSentIsScopedToItsRule — a send for another rule must not answer for +// this one, or the repeat interval is clocked off somebody else's alarm. +func TestLastSentIsScopedToItsRule(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + at := time.UnixMilli(1_700_000_000_000).UTC() + + if err := s.MarkSent(ctx, "other_rule", at); err != nil { + t.Fatalf("mark sent: %v", err) + } + last, err := s.LastSent(ctx, "service_down") + if err != nil { + t.Fatalf("last sent: %v", err) + } + if !last.IsZero() { + t.Fatalf("want zero for a rule with no sends, got %v", last) + } + + if err := s.MarkSent(ctx, "service_down", at); err != nil { + t.Fatalf("mark sent: %v", err) + } + last, err = s.LastSent(ctx, "service_down") + if err != nil { + t.Fatalf("last sent: %v", err) + } + if !last.Equal(at) { + t.Fatalf("want %v, got %v", at, last) + } +} diff --git a/internal/store/digest.go b/internal/store/digest.go index 457b50e..1f53e3e 100644 --- a/internal/store/digest.go +++ b/internal/store/digest.go @@ -3,7 +3,9 @@ package store import ( "context" "crypto/sha256" + "database/sql" "encoding/hex" + "errors" "fmt" "time" ) @@ -44,19 +46,33 @@ func DigestBodyHash(rule, body string) string { } // EnqueueDigestEntry durably records a suppressed care candidate worth -// resurfacing later. If a pending entry with the same rule+body already +// resurfacing later. If a LIVE pending entry with the same rule+body already // exists, this is a no-op that returns the existing id and deduped=true — // the same suppressed nudge repeating across ticks must not pile up into // several copies of itself in the eventual bundle. +// +// "Live" carries the same expiry test PendingDigestEntries reads with, and for +// the same reason: a row past its expires_ts is still status='pending' until +// the sweep gets to it, and the tick enqueues before it sweeps. Deduping +// against one meant reporting deduped=true against an entry that will never be +// spoken — the caller drops the phrasing it just paid the LLM for and nothing +// reaches the bundle. Not yet swept must not mean still deliverable on the +// write side either. func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity int, body string, now, expiresAt time.Time) (id int64, deduped bool, err error) { hash := DigestBodyHash(rule, body) var existing int64 err = s.db.QueryRowContext(ctx, - `SELECT id FROM digest_entries WHERE status = ? AND rule = ? AND body_hash = ? LIMIT 1`, - DigestPending, rule, hash).Scan(&existing) + `SELECT id FROM digest_entries + WHERE status = ? AND rule = ? AND body_hash = ? AND expires_ts > ? LIMIT 1`, + DigestPending, rule, hash, now.UnixMilli()).Scan(&existing) if err == nil { return existing, true, nil } + if !errors.Is(err, sql.ErrNoRows) { + // A real read failure is not "nothing there". Inserting anyway would + // duplicate an entry whose existence we never established. + return 0, false, fmt.Errorf("enqueue digest entry: dedupe lookup: %w", err) + } res, err := s.db.ExecContext(ctx, `INSERT INTO digest_entries (rule, severity, body, body_hash, status, created_ts, expires_ts) diff --git a/internal/store/digest_test.go b/internal/store/digest_test.go index 9c78392..c100e3e 100644 --- a/internal/store/digest_test.go +++ b/internal/store/digest_test.go @@ -200,3 +200,46 @@ func TestDigestEntryDrainMarksDrainedNotDeleted(t *testing.T) { } } } + +// TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry — an entry past its +// expires_ts is still status='pending' until the sweep runs, and the tick +// enqueues before it sweeps. Deduping against one reports deduped=true for a +// row PendingDigestEntries will never hand back, so the suppressed nudge is +// dropped instead of held: the entry says the thing was recorded when nothing +// was. +func TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + created := time.Now() + expiresAt := created.Add(time.Hour) + + first, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", created, expiresAt) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + if deduped { + t.Fatal("first enqueue must not report deduped") + } + + // A tick after the expiry, with the sweep not yet run: the same suppressed + // nudge comes round again and must be recorded afresh. + after := expiresAt.Add(time.Minute) + second, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", after, after.Add(time.Hour)) + if err != nil { + t.Fatalf("re-enqueue: %v", err) + } + if deduped { + t.Fatal("an expired entry must not swallow a fresh one") + } + if second == first { + t.Fatalf("want a new row, got the expired one back: id=%d", second) + } + + entries, err := s.PendingDigestEntries(ctx, after) + if err != nil { + t.Fatalf("pending: %v", err) + } + if len(entries) != 1 || entries[0].ID != second { + t.Fatalf("want the fresh entry %d pending, got %+v", second, entries) + } +}