feat: add read-only retention inventory
This commit is contained in:
@@ -98,4 +98,15 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
|
||||
if _, err := ReadOperationalSnapshot(cancelled, db); err == nil {
|
||||
t.Fatal("cancelled context unexpectedly completed snapshot")
|
||||
}
|
||||
f.exec("UPDATE pulse_tasks SET completed_at=DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 2 HOUR), updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?", f.resources.NamespaceID, task.ID)
|
||||
inventory, err := ReadRetentionInventory(ctx, db, time.Now().UTC().Add(-time.Hour))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if inventory.Cutoff.IsZero() || inventory.TerminalTasks == 0 || inventory.UncertainTasks != 0 {
|
||||
t.Fatalf("retention inventory=%+v", inventory)
|
||||
}
|
||||
if _, err := ReadRetentionInventory(ctx, db, time.Now().UTC().Add(time.Hour)); err == nil {
|
||||
t.Fatal("future retention cutoff unexpectedly accepted")
|
||||
}
|
||||
}
|
||||
|
||||
83
internal/storage/mysql/retention_inventory.go
Normal file
83
internal/storage/mysql/retention_inventory.go
Normal file
@@ -0,0 +1,83 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
)
|
||||
|
||||
// RetentionInventory 是指定截止时间前的只读保留清单。
|
||||
// 它只提供核对数量,不代表记录已经满足归档条件,也不执行删除或搬迁。
|
||||
type RetentionInventory struct {
|
||||
// Cutoff 是清单使用的 UTC 截止时间,严格早于该时间的记录才会计入。
|
||||
Cutoff time.Time `json:"cutoff"`
|
||||
// ExpiredIdempotency 是已过幂等保留期的请求记录数量。
|
||||
ExpiredIdempotency uint64 `json:"expired_idempotency"`
|
||||
// TerminalTasks 是已进入确定终态且已完成的任务数量。
|
||||
TerminalTasks uint64 `json:"terminal_tasks"`
|
||||
// UncertainTasks 是仍带未知效果的任务数量,必须人工核对后才能处理。
|
||||
UncertainTasks uint64 `json:"uncertain_tasks"`
|
||||
// ActiveTasks 是尚未终结的任务数量,不能进入普通历史归档。
|
||||
ActiveTasks uint64 `json:"active_tasks"`
|
||||
// TerminalWorkflows 是已进入确定终态且已完成的流程数量。
|
||||
TerminalWorkflows uint64 `json:"terminal_workflows"`
|
||||
// ActiveWorkflows 是尚未终结的流程数量,不能进入普通历史归档。
|
||||
ActiveWorkflows uint64 `json:"active_workflows"`
|
||||
// AcknowledgedDeliveries 是已经确认的结果交付数量。
|
||||
AcknowledgedDeliveries uint64 `json:"acknowledged_deliveries"`
|
||||
// UnconfirmedDeliveries 是仍未确认的结果交付数量,不能无声清理。
|
||||
UnconfirmedDeliveries uint64 `json:"unconfirmed_deliveries"`
|
||||
// ProcessedInbox 是已经完成内部消费的流程事件数量。
|
||||
ProcessedInbox uint64 `json:"processed_inbox"`
|
||||
// UnprocessedInbox 是尚未完成内部消费的流程事件数量,不能清理。
|
||||
UnprocessedInbox uint64 `json:"unprocessed_inbox"`
|
||||
// Events 是截止时间前追加的不可变事件数量。
|
||||
Events uint64 `json:"events"`
|
||||
// AuditEvents 是截止时间前追加的管理审计数量,默认应单独保留。
|
||||
AuditEvents uint64 `json:"audit_events"`
|
||||
}
|
||||
|
||||
// ReadRetentionInventory performs bounded-by-time, read-only count queries.
|
||||
// It deliberately does not decide policy or mutate any table; callers must
|
||||
// review retention windows, backups, outstanding effects and audit rules first.
|
||||
func ReadRetentionInventory(ctx context.Context, db *sql.DB, cutoff time.Time) (RetentionInventory, error) {
|
||||
if db == nil {
|
||||
return RetentionInventory{}, errors.New("database is required")
|
||||
}
|
||||
if cutoff.IsZero() {
|
||||
return RetentionInventory{}, errors.New("cutoff is required")
|
||||
}
|
||||
cutoff = cutoff.UTC()
|
||||
var now time.Time
|
||||
if err := db.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now); err != nil {
|
||||
return RetentionInventory{}, errors.New("database time unavailable")
|
||||
}
|
||||
if cutoff.After(now.UTC()) {
|
||||
return RetentionInventory{}, errors.New("cutoff cannot be in the future")
|
||||
}
|
||||
result := RetentionInventory{Cutoff: cutoff}
|
||||
queries := []struct {
|
||||
destination *uint64
|
||||
query string
|
||||
}{
|
||||
{&result.ExpiredIdempotency, "SELECT COUNT(*) FROM orbit_idempotency_keys WHERE expires_at<?"},
|
||||
{&result.TerminalTasks, "SELECT COUNT(*) FROM pulse_tasks WHERE state IN ('SUCCEEDED','FAILED','CANCELLED') AND completed_at IS NOT NULL AND completed_at<? AND effect_uncertain=FALSE"},
|
||||
{&result.UncertainTasks, "SELECT COUNT(*) FROM pulse_tasks WHERE effect_uncertain=TRUE AND created_at<?"},
|
||||
{&result.ActiveTasks, "SELECT COUNT(*) FROM pulse_tasks WHERE state NOT IN ('SUCCEEDED','FAILED','CANCELLED') AND created_at<?"},
|
||||
{&result.TerminalWorkflows, "SELECT COUNT(*) FROM weave_instances WHERE state IN ('SUCCEEDED','FAILED','CANCELLED') AND completed_at IS NOT NULL AND completed_at<?"},
|
||||
{&result.ActiveWorkflows, "SELECT COUNT(*) FROM weave_instances WHERE state NOT IN ('SUCCEEDED','FAILED','CANCELLED') AND created_at<?"},
|
||||
{&result.AcknowledgedDeliveries, "SELECT COUNT(*) FROM orbit_deliveries WHERE state='ACKED' AND acknowledged_at IS NOT NULL AND acknowledged_at<?"},
|
||||
{&result.UnconfirmedDeliveries, "SELECT COUNT(*) FROM orbit_deliveries WHERE state<>'ACKED' AND created_at<?"},
|
||||
{&result.ProcessedInbox, "SELECT COUNT(*) FROM weave_inbox WHERE state='PROCESSED' AND processed_at IS NOT NULL AND processed_at<?"},
|
||||
{&result.UnprocessedInbox, "SELECT COUNT(*) FROM weave_inbox WHERE state<>'PROCESSED' AND created_at<?"},
|
||||
{&result.Events, "SELECT COUNT(*) FROM orbit_events WHERE occurred_at<?"},
|
||||
{&result.AuditEvents, "SELECT COUNT(*) FROM orbit_audit_events WHERE created_at<?"},
|
||||
}
|
||||
for _, item := range queries {
|
||||
if err := db.QueryRowContext(ctx, item.query, cutoff).Scan(item.destination); err != nil {
|
||||
return RetentionInventory{}, errors.New("retention inventory unavailable")
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
Reference in New Issue
Block a user