retention: write the audit row before the actions_log cap
RetainWithAudit appends the 'retain' audit row inside the same transaction and before the cap, so the cap counts it. After a run, actions_log holds at most EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS rows, including the audit row. Adds a test. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
@@ -344,16 +344,14 @@ func runRetain(cfg *config.Config, logger *slog.Logger) error {
|
|||||||
BodyMaxAge: time.Duration(cfg.RetentionDays) * 24 * time.Hour,
|
BodyMaxAge: time.Duration(cfg.RetentionDays) * 24 * time.Hour,
|
||||||
ActionsLogMaxRows: cfg.ActionsLogMaxRows,
|
ActionsLogMaxRows: cfg.ActionsLogMaxRows,
|
||||||
}
|
}
|
||||||
res, err := st.Retain(ctx, policy)
|
// The audit row is written inside the same transaction, before the cap, so
|
||||||
|
// the cap counts it. This is what keeps actions_log bounded across runs.
|
||||||
|
detail := fmt.Sprintf("bodies_cleared=%%d retention_days=%d max_rows=%d",
|
||||||
|
cfg.RetentionDays, cfg.ActionsLogMaxRows)
|
||||||
|
res, err := st.RetainWithAudit(ctx, policy, "scheduler", detail)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := st.LogAction(ctx, "retain", "scheduler",
|
|
||||||
fmt.Sprintf("bodies_cleared=%d actions_deleted=%d retention_days=%d max_rows=%d",
|
|
||||||
res.BodiesCleared, res.ActionsDeleted, cfg.RetentionDays, cfg.ActionsLogMaxRows)); err != nil {
|
|
||||||
// Retention already committed; a failed audit row must not fail the job.
|
|
||||||
logger.Error("log retention action", "error", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
cutoff := "disabled"
|
cutoff := "disabled"
|
||||||
if !res.BodyCutoff.IsZero() {
|
if !res.BodyCutoff.IsZero() {
|
||||||
|
|||||||
@@ -49,6 +49,14 @@ type RetentionResult struct {
|
|||||||
// pass never leaves the database in a mixed state. It is safe to run on a
|
// pass never leaves the database in a mixed state. It is safe to run on a
|
||||||
// schedule: every statement is idempotent.
|
// schedule: every statement is idempotent.
|
||||||
func (s *Store) Retain(ctx context.Context, p RetentionPolicy) (RetentionResult, error) {
|
func (s *Store) Retain(ctx context.Context, p RetentionPolicy) (RetentionResult, error) {
|
||||||
|
return s.RetainWithAudit(ctx, p, "", "")
|
||||||
|
}
|
||||||
|
|
||||||
|
// RetainWithAudit is Retain plus one actions_log row (FR-13), appended inside
|
||||||
|
// the same transaction before the cap is applied so the audit row is itself
|
||||||
|
// counted by the cap. An empty detail writes no audit row. This keeps the
|
||||||
|
// action log bounded even though every run records that it ran.
|
||||||
|
func (s *Store) RetainWithAudit(ctx context.Context, p RetentionPolicy, actor, detail string) (RetentionResult, error) {
|
||||||
var res RetentionResult
|
var res RetentionResult
|
||||||
|
|
||||||
tx, err := s.db.BeginTx(ctx, nil)
|
tx, err := s.db.BeginTx(ctx, nil)
|
||||||
@@ -74,6 +82,14 @@ WHERE body_text IS NOT NULL AND body_text <> '' AND received_at < ?`,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if detail != "" {
|
||||||
|
if _, err := tx.ExecContext(ctx, `
|
||||||
|
INSERT INTO actions_log (message_id, action, actor, detail)
|
||||||
|
VALUES (NULL, 'retain', ?, ?)`, nullString(actor), detail); err != nil {
|
||||||
|
return res, fmt.Errorf("log retention action: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if p.ActionsLogMaxRows > 0 {
|
if p.ActionsLogMaxRows > 0 {
|
||||||
// Keep the newest N rows and delete the rest. id is monotonic, so it
|
// Keep the newest N rows and delete the rest. id is monotonic, so it
|
||||||
// is the stable ordering for an append-only table.
|
// is the stable ordering for an append-only table.
|
||||||
|
|||||||
@@ -121,6 +121,35 @@ func TestRetainCapsActionsLog(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRetainWithAuditCountsItsOwnRow(t *testing.T) {
|
||||||
|
s, ctx := testStore(t)
|
||||||
|
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
if err := s.LogAction(ctx, "fetch", "scheduler", "pass"); err != nil {
|
||||||
|
t.Fatalf("LogAction() %d error = %v", i, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err := s.RetainWithAudit(ctx, RetentionPolicy{ActionsLogMaxRows: 3}, "scheduler", "bodies_cleared=0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("RetainWithAudit() error = %v", err)
|
||||||
|
}
|
||||||
|
// 5 existing + 1 audit = 6; keep 3, so 3 are deleted.
|
||||||
|
if res.ActionsDeleted != 3 {
|
||||||
|
t.Errorf("ActionsDeleted = %d, want 3", res.ActionsDeleted)
|
||||||
|
}
|
||||||
|
entries, err := s.RecentActions(ctx, 100)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("RecentActions() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(entries) != 3 {
|
||||||
|
t.Fatalf("actions_log rows = %d, want 3 (cap counts the audit row)", len(entries))
|
||||||
|
}
|
||||||
|
if entries[0].Action != "retain" {
|
||||||
|
t.Errorf("newest row action = %q, want retain", entries[0].Action)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestRetainDisabledPolicyIsNoOp(t *testing.T) {
|
func TestRetainDisabledPolicyIsNoOp(t *testing.T) {
|
||||||
s, ctx := testStore(t)
|
s, ctx := testStore(t)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user