370 lines
16 KiB
Go
370 lines
16 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/base64"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
"git.yuntuo.hk/k2safe/orbit/internal/retention"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||
)
|
||
|
||
func TestIntegrationRetentionPolicyIsSharedAuditedAndIdempotent(t *testing.T) {
|
||
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
||
if dsn == "" {
|
||
t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration")
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||
defer cancel()
|
||
db, err := Open(ctx, dsn, 8, 2)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer db.Close()
|
||
if err := Migrate(ctx, db); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
var boot BootstrapResult
|
||
app := testID(t)
|
||
if err := Bootstrap(ctx, db, BootstrapInput{
|
||
Application: string(app), Namespace: "default", Environment: "test", Queue: "admin",
|
||
ConsoleUsername: "admin-" + string(app), ConsolePassword: "a-strong-password-123",
|
||
}, func(v BootstrapResult) error {
|
||
boot = v
|
||
return nil
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
login, err := NewConsoleAuthStore(db).Login(ctx, "admin-"+string(app), "a-strong-password-123", "retention-test", "127.0.0.1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
token, err := auth.Parse(login.Token)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defaults := retention.Defaults(boot.NamespaceID)
|
||
store := NewRetentionPolicyStore(db, defaults)
|
||
got, err := store.GetRetentionPolicy(ctx, boot.NamespaceID)
|
||
if err != nil || got.Source != "default" || got.Revision != 0 || got.TaskHistoryDays != 365 {
|
||
t.Fatalf("default policy: %+v %v", got, err)
|
||
}
|
||
|
||
in := retention.UpdateInput{
|
||
ExpectedRevision: 0, Enabled: true, IdempotencyDays: 120,
|
||
TaskHistoryDays: 730, WorkflowHistoryDays: 730, EventLogDays: 365,
|
||
DeliveryHistoryDays: 120, InboxHistoryDays: 120, AuditLogDays: 2555,
|
||
Reason: "production retention policy review",
|
||
}
|
||
updated, err := store.UpdateRetentionPolicy(ctx, token, "retention-change-1", in)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if updated.Source != "database" || updated.Revision != 1 || !updated.Enabled || updated.TaskHistoryDays != 730 {
|
||
t.Fatalf("updated policy: %+v", updated)
|
||
}
|
||
// A second store represents another Orbit node; it must read the same MySQL policy.
|
||
other := NewRetentionPolicyStore(db, retention.Defaults(boot.NamespaceID))
|
||
shared, err := other.GetRetentionPolicy(ctx, boot.NamespaceID)
|
||
if err != nil || shared.Revision != 1 || shared.TaskHistoryDays != 730 || !shared.Enabled {
|
||
t.Fatalf("cross-node policy read: %+v %v", shared, err)
|
||
}
|
||
replayed, err := store.UpdateRetentionPolicy(ctx, token, "retention-change-1", in)
|
||
if err != nil || replayed.Revision != updated.Revision || replayed.TaskHistoryDays != updated.TaskHistoryDays {
|
||
t.Fatalf("idempotent replay: %+v %v", replayed, err)
|
||
}
|
||
conflict := in
|
||
conflict.ExpectedRevision = 0
|
||
conflict.TaskHistoryDays = 900
|
||
if _, err := store.UpdateRetentionPolicy(ctx, token, "retention-change-2", conflict); faultCode(err) != "STATE_CONFLICT" {
|
||
t.Fatalf("stale revision accepted: %v", err)
|
||
}
|
||
var audits int
|
||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND action='retention.policy.update' AND resource_kind='retention_policy'`, boot.NamespaceID).Scan(&audits); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if audits != 1 {
|
||
t.Fatalf("audit rows=%d want 1 after idempotent replay", audits)
|
||
}
|
||
var actor, reason string
|
||
if err := db.QueryRowContext(ctx, "SELECT actor_id,reason FROM orbit_audit_events WHERE namespace_id=? AND action='retention.policy.update'", boot.NamespaceID).Scan(&actor, &reason); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if actor == "" || reason != in.Reason || actor == "mysql:database" {
|
||
t.Fatalf("unexpected retention audit identity: actor=%q reason=%q", actor, reason)
|
||
}
|
||
t.Logf("retention policy revision %d is shared across stores, audited once, and safe on replay", updated.Revision)
|
||
}
|
||
|
||
func faultCode(err error) string {
|
||
if err == nil {
|
||
return ""
|
||
}
|
||
return fault.Code(err)
|
||
}
|
||
|
||
func TestIntegrationRetentionDiskGrowthAndPreview(t *testing.T) {
|
||
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
||
if dsn == "" {
|
||
t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration")
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||
defer cancel()
|
||
db, err := Open(ctx, dsn, 12, 4)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer db.Close()
|
||
if err := Migrate(ctx, db); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var boot BootstrapResult
|
||
app := testID(t)
|
||
if err := Bootstrap(ctx, db, BootstrapInput{
|
||
Application: string(app), Namespace: "growth", Environment: "test", Queue: "growth",
|
||
ConsoleUsername: "growth-admin-" + string(app), ConsolePassword: "a-strong-password-123",
|
||
}, func(v BootstrapResult) error { boot = v; return nil }); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
apiToken, err := auth.Parse(boot.Token)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
login, err := NewConsoleAuthStore(db).Login(ctx, "growth-admin-"+string(app), "a-strong-password-123", "retention-growth-test", "127.0.0.1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
consoleToken, err := auth.Parse(login.Token)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := db.ExecContext(ctx, "ANALYZE TABLE pulse_tasks, orbit_events, orbit_audit_events"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
before := retentionTableBytes(t, ctx, db)
|
||
serviceStore := service.New(NewTaskStore(db))
|
||
// 256 rows × 32 KiB is deliberately larger than one InnoDB allocation page,
|
||
// so the information_schema sample must observe physical table growth.
|
||
payload := json.RawMessage(`{"blob":"` + strings.Repeat("x", 32768) + `"}`)
|
||
ids := make([]string, 0, 256)
|
||
for i := 0; i < 256; i++ {
|
||
out, err := serviceStore.Submit(ctx, apiToken, "growth-"+string(app)+"-"+string(rune('a'+i%26))+"-"+string(testID(t)), service.SubmitInput{
|
||
QueueID: boot.QueueID, Handler: "retention.growth", HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload,
|
||
TaskTimeoutMS: 60000, RecoverySafety: pulse.RecoveryReconcileOnly, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 30000},
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ids = append(ids, string(out.Task.ID))
|
||
}
|
||
completedAt := time.Now().UTC().Add(-48 * time.Hour).Truncate(time.Microsecond)
|
||
for _, id := range ids {
|
||
if _, err := db.ExecContext(ctx, `UPDATE pulse_tasks SET state='FAILED',completed_at=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=UNHEX(?)`, completedAt, boot.NamespaceID, id); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
// 同一完成时间验证复合游标；未知效果和活跃租约不能因时间足够旧而被导出。
|
||
if _, err := db.ExecContext(ctx, "UPDATE pulse_tasks SET state='NEEDS_REVIEW',completed_at=NULL,effect_uncertain=TRUE WHERE namespace_id=? AND id=UNHEX(?)", boot.NamespaceID, ids[0]); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := db.ExecContext(ctx, "UPDATE pulse_tasks SET state='RUNNING',completed_at=NULL,current_attempt_id=UNHEX('00000000000000000000000000000001'),lease_until=UTC_TIMESTAMP(6)+INTERVAL 1 HOUR WHERE namespace_id=? AND id=UNHEX(?)", boot.NamespaceID, ids[1]); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := db.ExecContext(ctx, "ANALYZE TABLE pulse_tasks, orbit_events, orbit_audit_events"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
after := retentionTableBytes(t, ctx, db)
|
||
if after <= before {
|
||
t.Fatalf("retention table bytes did not grow during bounded growth sample: before=%d after=%d", before, after)
|
||
}
|
||
store := NewRetentionPolicyStore(db, retention.Defaults(boot.NamespaceID))
|
||
_, err = store.UpdateRetentionPolicy(ctx, consoleToken, "growth-retention-policy", retention.UpdateInput{
|
||
ExpectedRevision: 0, Enabled: true, IdempotencyDays: 90, TaskHistoryDays: 1,
|
||
WorkflowHistoryDays: 30, EventLogDays: 30, DeliveryHistoryDays: 30, InboxHistoryDays: 30,
|
||
AuditLogDays: 365, Reason: "bounded disk growth rehearsal",
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
preview, err := store.PreviewRetention(ctx, consoleToken)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var taskSample retention.CandidatePage
|
||
for _, item := range preview.Items {
|
||
if item.Kind == "tasks" {
|
||
taskSample = item
|
||
}
|
||
}
|
||
if taskSample.Count == 0 {
|
||
t.Fatalf("backdated terminal tasks were not visible as retention candidates: %+v", preview.Items)
|
||
}
|
||
snapshot, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", Limit: 1, ExpectedRevision: 1})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if snapshot.FormatVersion != 1 || snapshot.Content == "" {
|
||
t.Fatalf("invalid retention snapshot envelope: %+v", snapshot)
|
||
}
|
||
sum := sha256.Sum256([]byte(snapshot.Content))
|
||
if snapshot.SHA256 != hex.EncodeToString(sum[:]) || !json.Valid([]byte(snapshot.Content)) || !strings.Contains(snapshot.Content, `"source_retained":true`) {
|
||
t.Fatalf("retention snapshot integrity/source marker failed: sha=%s content=%s", snapshot.SHA256, snapshot.Content[:min(len(snapshot.Content), 200)])
|
||
}
|
||
var first archiveData
|
||
if err := json.Unmarshal([]byte(snapshot.Content), &first); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if first.NextCursor == "" {
|
||
t.Fatal("first page lost continuation cursor")
|
||
}
|
||
seen := map[string]bool{}
|
||
page := first
|
||
batches := 0
|
||
for {
|
||
batches++
|
||
if page.Cutoff != first.Cutoff || page.Policy.Revision != 1 || !page.SourceRetained {
|
||
t.Fatalf("batch changed boundary: %+v", page.Policy)
|
||
}
|
||
if page.RootCount != len(page.Tables["pulse_tasks"]) {
|
||
t.Fatal("root count mismatch")
|
||
}
|
||
for _, row := range page.Tables["pulse_tasks"] {
|
||
id, ok := row["id"].(string)
|
||
if !ok || seen[id] || id == ids[0] || id == ids[1] {
|
||
t.Fatalf("duplicate/unsafe root: %v", row["id"])
|
||
}
|
||
seen[id] = true
|
||
}
|
||
if page.NextCursor == "" {
|
||
break
|
||
}
|
||
if batches > 10 {
|
||
t.Fatal("cursor did not make bounded progress")
|
||
}
|
||
next, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", Limit: 37, Cursor: page.NextCursor, ExpectedRevision: 1})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
digest := sha256.Sum256([]byte(next.Content))
|
||
if next.SHA256 != hex.EncodeToString(digest[:]) {
|
||
t.Fatal("batch digest mismatch")
|
||
}
|
||
if err := json.Unmarshal([]byte(next.Content), &page); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
if len(seen) != len(ids)-2 {
|
||
t.Fatalf("exported %d roots; want %d", len(seen), len(ids)-2)
|
||
}
|
||
var retained int
|
||
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?", boot.NamespaceID).Scan(&retained); err != nil || retained != len(ids) {
|
||
t.Fatalf("source data changed: count=%d err=%v", retained, err)
|
||
}
|
||
for _, in := range []retention.ArchiveInput{
|
||
{Kind: "audit", Cursor: first.NextCursor, ExpectedRevision: 1},
|
||
{Kind: "tasks", Cursor: "not-a-cursor", ExpectedRevision: 1},
|
||
{Kind: "tasks", Limit: 101, ExpectedRevision: 1},
|
||
} {
|
||
if _, err := store.ExportRetention(ctx, consoleToken, in); err == nil {
|
||
t.Fatalf("invalid export accepted: %+v", in)
|
||
}
|
||
}
|
||
raw, err := base64.RawURLEncoding.DecodeString(first.NextCursor)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var foreign archiveCursor
|
||
if err := json.Unmarshal(raw, &foreign); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
foreign.Namespace = testID(t)
|
||
raw, err = json.Marshal(foreign)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 1, Cursor: base64.RawURLEncoding.EncodeToString(raw)}); err == nil {
|
||
t.Fatal("foreign namespace cursor accepted")
|
||
}
|
||
if _, err := store.ExportRetention(ctx, apiToken, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 1}); err == nil {
|
||
t.Fatal("business bearer token accepted as console admin")
|
||
}
|
||
|
||
jobSpec := retention.JobSpec{NamespaceID: boot.NamespaceID, Kind: "tasks", PolicyRevision: 1, BatchSize: 37}
|
||
jobDir := filepath.Join(t.TempDir(), "retention-job")
|
||
jobOptions := retention.JobOptions{MaxBatches: 1, MaxBytes: 64 << 20}
|
||
exportJob := func(ctx context.Context, job core.ID, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) {
|
||
return store.ExportRetentionForJob(ctx, boot.NamespaceID, job, in)
|
||
}
|
||
firstJob, err := retention.RunArchive(ctx, jobDir, jobSpec, jobOptions, exportJob)
|
||
if err != nil || firstJob.Completed || firstJob.Roots != 37 {
|
||
t.Fatalf("archive job first batch=%+v err=%v", firstJob, err)
|
||
}
|
||
jobOptions.MaxBatches = 10
|
||
finishedJob, err := retention.RunArchive(ctx, jobDir, jobSpec, jobOptions, exportJob)
|
||
if err != nil || !finishedJob.Completed || finishedJob.Roots != 254 || finishedJob.Batches != 7 || finishedJob.JobID != firstJob.JobID {
|
||
t.Fatalf("archive job resumed=%+v err=%v", finishedJob, err)
|
||
}
|
||
verifiedJob, err := retention.VerifyArchive(ctx, jobDir)
|
||
if err != nil || verifiedJob != finishedJob {
|
||
t.Fatalf("offline job verification=%+v err=%v", verifiedJob, err)
|
||
}
|
||
var jobAudits int
|
||
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND actor_id=? AND action='retention.archive.export'", boot.NamespaceID, "archive:"+string(finishedJob.JobID)).Scan(&jobAudits); err != nil || jobAudits != 7 {
|
||
t.Fatalf("job audit count=%d err=%v", jobAudits, err)
|
||
}
|
||
if _, err := db.ExecContext(ctx, "UPDATE orbit_namespaces SET enabled=FALSE WHERE id=?", boot.NamespaceID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.ExportRetentionForJob(ctx, boot.NamespaceID, finishedJob.JobID, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 1}); faultCode(err) != "PERMISSION_DENIED" {
|
||
t.Fatalf("disabled namespace exported: %v", err)
|
||
}
|
||
if _, err := db.ExecContext(ctx, "UPDATE orbit_namespaces SET enabled=TRUE WHERE id=?", boot.NamespaceID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Logf("durable archive job: roots=%d batches=%d audit_rows=%d offline_verified=true", finishedJob.Roots, finishedJob.Batches, jobAudits)
|
||
policyChange := retention.UpdateInput{ExpectedRevision: 1, Enabled: true, IdempotencyDays: 90, TaskHistoryDays: 2, WorkflowHistoryDays: 30, EventLogDays: 30, DeliveryHistoryDays: 30, InboxHistoryDays: 30, AuditLogDays: 365, Reason: "verify frozen cursor boundary"}
|
||
if _, err := store.UpdateRetentionPolicy(ctx, consoleToken, "growth-retention-revise", policyChange); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 1, Cursor: first.NextCursor}); faultCode(err) != "STATE_CONFLICT" {
|
||
t.Fatalf("stale policy allowed: %v", err)
|
||
}
|
||
if _, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 2, Cursor: first.NextCursor}); err == nil {
|
||
t.Fatal("old cursor accepted with new policy")
|
||
}
|
||
policyChange.ExpectedRevision, policyChange.Enabled = 2, false
|
||
if _, err := store.UpdateRetentionPolicy(ctx, consoleToken, "growth-retention-disable", policyChange); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 3}); faultCode(err) != "STATE_CONFLICT" {
|
||
t.Fatalf("disabled policy allowed: %v", err)
|
||
}
|
||
t.Logf("archive pagination: batches=%d unique_roots=%d protected_roots=2 source_rows=%d; digest, namespace, kind, revision, disabled policy and console-only checks passed", batches, len(seen), retained)
|
||
t.Logf("bounded retention growth sample inserted %d tasks (254 terminal, 2 protected); bytes before=%d after=%d delta=%d; preview tasks=%d more=%v", len(ids), before, after, after-before, taskSample.Count, taskSample.More)
|
||
}
|
||
|
||
func retentionTableBytes(t *testing.T, ctx context.Context, db interface {
|
||
QueryRowContext(context.Context, string, ...any) *sql.Row
|
||
}) int64 {
|
||
t.Helper()
|
||
var bytes int64
|
||
if err := db.QueryRowContext(ctx, `SELECT COALESCE(SUM(data_length+index_length),0) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('pulse_tasks','orbit_events','orbit_audit_events')`).Scan(&bytes); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return bytes
|
||
}
|