From e301f6526d75858e9ed10ffb5866dba698386c97 Mon Sep 17 00:00:00 2001 From: Sami Salih Ibrahimbas Date: Sat, 12 Sep 2026 19:20:21 +0300 Subject: [PATCH 1/2] docs: link hosted Trail Explorer demo --- README.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/README.md b/README.md index 10e599d..b48b6f2 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,13 @@ It makes structured operational events nearly as easy to emit as `fmt.Println`, while preserving enough identity to reconstruct a business flow, entity history, or execution later in the sink of your choice. +## Live demo + +Explore a deterministic one-million-event Trail dataset at +**[trail.vestavision.io](https://trail.vestavision.io)**. The hosted Explorer +demonstrates flow timelines, execution relationships, entity history, HTTP +events, structured fields, and payload references using synthetic demo data. + ```go trail.Init(trail.Config{ Service: "order-worker", From 9dd228f59cee3984db4c84cc83d8deee68e722bd Mon Sep 17 00:00:00 2001 From: Sami Salih Ibrahimbas Date: Sat, 12 Sep 2026 21:32:34 +0300 Subject: [PATCH 2/2] feat: add archive-backed retention --- README.md | 27 ++ RETENTION.md | 154 ++++++++ archive/archive.go | 317 ++++++++++++++++ archive/archive_test.go | 111 ++++++ cmd/trail-explorer-api/main.go | 36 +- cmd/trail-retention/main.go | 362 ++++++++++++++++++ cmd/trail-retention/main_test.go | 36 ++ convention/convention.go | 7 + explorerapi/server.go | 115 ++++++ retention.example.json | 29 ++ retention/retention.go | 548 +++++++++++++++++++++++++++ retention/retention_test.go | 325 ++++++++++++++++ storage/clickhouse/migrations.go | 14 + storage/clickhouse/retention.go | 530 ++++++++++++++++++++++++++ storage/clickhouse/retention_test.go | 36 ++ storage/normalize.go | 2 + storage/normalize_test.go | 3 +- storage/postgres/migrations.go | 11 + storage/postgres/retention.go | 336 ++++++++++++++++ storage/postgres/retention_test.go | 36 ++ storage/storage.go | 1 + 21 files changed, 3033 insertions(+), 3 deletions(-) create mode 100644 RETENTION.md create mode 100644 archive/archive.go create mode 100644 archive/archive_test.go create mode 100644 cmd/trail-retention/main.go create mode 100644 cmd/trail-retention/main_test.go create mode 100644 retention.example.json create mode 100644 retention/retention.go create mode 100644 retention/retention_test.go create mode 100644 storage/clickhouse/retention.go create mode 100644 storage/clickhouse/retention_test.go create mode 100644 storage/postgres/retention.go create mode 100644 storage/postgres/retention_test.go diff --git a/README.md b/README.md index b48b6f2..ee79043 100644 --- a/README.md +++ b/README.md @@ -281,6 +281,33 @@ profile and point both Go services at `TRAIL_POSTGRES_URL`. See [PROVING_GROUND.md](PROVING_GROUND.md) for reproducible validation tiers and the measurement checklist. +## Retention + +Retention is an optional lifecycle component and is completely separate from +`trail.Init`, `trail.Log`, and `trail.Close`. Trail never starts cleanup +goroutines, opens retention database connections, or runs a scheduler as part +of the logging library. + +The one-shot `trail-retention` command applies explicit ordered policies to +ClickHouse or PostgreSQL, archives every eligible batch, and only then deletes +live rows and safely garbage-collects unreferenced payloads: + +```bash +trail-retention run --config retention.json --dry-run +TRAIL_ARCHIVE_FILESYSTEM_ROOT=./trail-archives trail-retention run --config retention.json +``` + +There is no destructive default policy and no delete-without-archive mode. A +failed archive write, verification, or catalog commit leaves live events +untouched. Archives use versioned gzip NDJSON plus a verified manifest and +support bounded, idempotent restore. PostgreSQL uses bounded transactional +deletes; ClickHouse uses bounded synchronous mutations and reclaims +physical space during later merges. Payload GC is eventually consistent and +honors `payload.Ref.RetainUntil`, `RetentionClass`, shared content-addressed +references, and configured limits. See [RETENTION.md](RETENTION.md) for policy +precedence, configuration, and operational guidance; start from +[`retention.example.json`](retention.example.json). + ## Installation and development ```bash diff --git a/RETENTION.md b/RETENTION.md new file mode 100644 index 0000000..7266449 --- /dev/null +++ b/RETENTION.md @@ -0,0 +1,154 @@ +# Trail retention + +Retention is an optional, one-shot lifecycle operation. It is not initialized +by `trail.Init`, it does not run a scheduler, and it adds no work to Trail's +logging hot path. Run it from cron, a Kubernetes CronJob, Nomad, or manually. + +No implicit retention period exists. The command refuses to run without an +event policy, event class, or explicit default event TTL. Retention also cannot +be constructed or run without an archive store. Trail writes and verifies a +versioned gzip-NDJSON bundle, commits its manifest to the event store catalog, +and only then permits deletion. There is no delete-without-archive mode. + +## Configuration + +Durations use Go duration syntax. Policies are checked in order and the first +match wins. Empty match properties are wildcards. + +```json +{ + "event_policies": [ + {"name":"http-success", "match":{"kind":"provider.http", "http_success":true}, "ttl":"168h"}, + {"name":"http-error", "match":{"kind":"provider.http", "http_success":false}, "ttl":"720h"}, + {"name":"business", "match":{}, "ttl":"2160h"} + ], + "event_classes": {"audit":"8760h"}, + "payload_classes": {"http":"168h", "error":"720h", "audit":"8760h"}, + "default_payload_ttl":"168h", + "event_batch_size":1000, + "payload_batch_size":100, + "payload_concurrency":1, + "max_events":100000, + "max_payloads":10000, + "max_runtime":"10m", + "payload_grace_period":"24h" +} +``` + +An event class can be attached with `convention.RetentionClass("audit")`. +Payload classes and absolute expiry already travel in `payload.Ref` as +`RetentionClass` and `RetainUntil`. + +Event precedence is event class, first matching policy, default TTL, then keep +forever. Payload precedence is absolute `RetainUntil`, known payload class, +first matching payload policy, default payload TTL, then keep forever. An +unknown non-empty class is retained, rather than silently falling through. + +## Running + +The database and payload environment variables are the same as the ingestor +and Explorer API: + +```bash +TRAIL_STORE=postgres \ +TRAIL_POSTGRES_URL='postgres://trail:trail@localhost/trail?sslmode=disable' \ +TRAIL_ARCHIVE_STORE=filesystem \ +TRAIL_ARCHIVE_FILESYSTEM_ROOT='./trail-archives' \ +trail-retention run --config retention.json --dry-run + +TRAIL_STORE=clickhouse \ +TRAIL_CLICKHOUSE_ADDR=localhost:9000 \ +TRAIL_ARCHIVE_STORE=s3 \ +TRAIL_ARCHIVE_S3_ENDPOINT=localhost:9000 \ +TRAIL_ARCHIVE_S3_BUCKET=trail-archives \ +trail-retention run --config retention.json +``` + +`--max-events`, `--max-payloads`, and `--max-runtime` can lower the configured +work budget for a manual or exploratory run. + +Dry run evaluates the same bounded pages but does not stage payload candidates, +delete events, or delete objects. The command prints a JSON run summary and +returns a non-zero status on an operational failure. + +## Archive and restore + +Each bounded deletion batch produces an immutable `trail.archive/v1` +manifest, a gzip-compressed NDJSON event object, and copies of every referenced +payload object. The manifest is written last and is the commit marker. A failed +event write, payload copy, verification, or catalog commit leaves live events +untouched. + +The archive destination is mandatory and should be independent from active +payload storage. It may use the same S3-compatible service with a separate +bucket or non-overlapping prefix. Archive objects have no Trail-managed expiry +in this version. + +```bash +trail-retention archives --limit 100 +trail-retention inspect --archive-id +trail-retention restore --archive-id --offset 0 --limit 1000 +``` + +Restore copies and verifies payloads before merging events. PostgreSQL uses its +event ID primary key; ClickHouse uses its replacing key and an archive-specific +deduplication identity. Repeating a restore is safe. Restore large archives in +bounded chunks using `next_offset` until `complete` is true. + +Explorer archive routes are available when archive storage is configured. The +`POST /api/v1/archives/{id}/restore` route additionally requires +`TRAIL_EXPLORER_ENABLE_ARCHIVE_RESTORE=true`. Trail does not own authentication; +deployments must protect restore through their existing application or reverse +proxy authorization. + +## Database behavior + +PostgreSQL records the verified archive mapping first, then stages payload +candidates and deletes each bounded event batch in a single transaction. Keep +autovacuum enabled and monitor dead tuples for large +retention workloads; Trail never runs `VACUUM` for you. + +ClickHouse verifies the durable archive mapping before staging payload +candidates and issuing a synchronous bounded mutation. Physical disk space is +reclaimed by later merges, so a +successful run represents logical deletion rather than immediate disk +reclamation. Each mutation also rebuilds only the affected flow, execution, +and terminal summary keys so Explorer results remain correct without turning +every UI request into a raw-event aggregation. Trail does not use a table TTL +because dynamic ordered policies can differ between rows in the same +partition. + +## Payload safety + +Payload objects are content addressed and may be shared. Event deletion first +creates a durable candidate. A candidate must pass its expiry, wait through the +configured grace period, and have no remaining reference in live events before +the object is deleted. Failures remain retryable; a missing object is treated +as an idempotent success. + +Filesystem deletion is confined to the configured root. S3-compatible deletion +is confined to the configured bucket and prefix. The retention engine only +uses references normalized from Trail events and never accepts Explorer input +or arbitrary object keys. + +Payload GC is intentionally eventually consistent. A failed event deletion +cannot trigger payload deletion, while partial ClickHouse work can leave a +safe stale candidate or reference that delays reclamation. + +The initial collector discovers orphans from references staged before event +deletion. Objects that were written successfully but never appeared in any +ingested Trail event are not enumerated automatically; retain them with the +payload backend's own conservative lifecycle policy until an inventory-based +collector is configured in a later release. + +## Operational guidance + +- Always run `--dry-run` after changing policies. +- Back up both archive storage and the database archive catalog; verified + restore needs both. +- Start with small batch and work limits, then increase them while observing + database latency, ClickHouse mutations, PostgreSQL dead tuples, and runtime. +- Run only one ClickHouse retention command at a time. +- Schedule externally; Trail deliberately has no loop or scheduler mode. +- Keep the grace period longer than the maximum expected ingestion/redelivery + delay. diff --git a/archive/archive.go b/archive/archive.go new file mode 100644 index 0000000..be198f2 --- /dev/null +++ b/archive/archive.go @@ -0,0 +1,317 @@ +// Package archive writes immutable, verifiable Trail retention bundles. +package archive + +import ( + "bufio" + "bytes" + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "io" + "time" + + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/storage" +) + +const Version = "trail.archive/v1" + +type PayloadEntry struct { + Source payload.Ref `json:"source"` + Archive payload.Ref `json:"archive"` +} +type Manifest struct { + Version string `json:"version"` + ID string `json:"id"` + CreatedAt time.Time `json:"created_at"` + EventCount int `json:"event_count"` + EventIDs []string `json:"event_ids"` + Events payload.Ref `json:"events"` + Payloads []PayloadEntry `json:"payloads,omitempty"` +} +type Receipt struct { + Manifest Manifest `json:"manifest"` + ManifestRef payload.Ref `json:"manifest_ref"` +} + +type Page struct { + Items []Receipt `json:"items"` + HasMore bool `json:"has_more"` +} + +// Catalog is the small database-backed archive discovery boundary. Object +// store keys are never accepted as archive identities. +type Catalog interface { + ListArchives(context.Context, int, int) (Page, error) + GetArchive(context.Context, string) (Receipt, error) +} + +type Writer interface { + Write(context.Context, []storage.EventRecord) (Receipt, error) +} + +// RestoreStore accepts a bounded archive chunk. Implementations must make +// repeated writes idempotent by event ID. +type RestoreStore interface { + RestoreArchiveEvents(context.Context, string, []storage.EventRecord) (int, error) +} + +type RestoreResult struct { + ArchiveID string `json:"archive_id"` + Restored int `json:"restored"` + Next int `json:"next_offset,omitempty"` + Complete bool `json:"complete"` +} + +type BundleWriter struct { + archive payload.Store + sources map[string]payload.Store + now func() time.Time +} + +func NewWriter(store payload.Store, sources map[string]payload.Store) (*BundleWriter, error) { + if store == nil { + return nil, errors.New("trail/archive: archive store is required") + } + return &BundleWriter{archive: store, sources: sources, now: func() time.Time { return time.Now().UTC() }}, nil +} + +func (w *BundleWriter) Write(ctx context.Context, events []storage.EventRecord) (Receipt, error) { + if len(events) == 0 { + return Receipt{}, errors.New("trail/archive: empty event batch") + } + var eventData bytes.Buffer + h := sha256.New() + mw := io.MultiWriter(&eventData, h) + bw := bufio.NewWriter(mw) + enc := json.NewEncoder(bw) + for i := range events { + if err := enc.Encode(events[i]); err != nil { + return Receipt{}, err + } + } + if err := bw.Flush(); err != nil { + return Receipt{}, err + } + eventRef, err := w.archive.Put(ctx, bytes.NewReader(eventData.Bytes()), payload.PutOptions{ContentType: "application/x-ndjson", Compression: payload.CompressionGZIP}) + if err != nil { + return Receipt{}, err + } + id := hex.EncodeToString(h.Sum(nil)) + manifest := Manifest{Version: Version, ID: id, CreatedAt: w.now(), EventCount: len(events), Events: eventRef, EventIDs: make([]string, len(events))} + for i := range events { + manifest.EventIDs[i] = events[i].ID.String() + } + seen := map[string]bool{} + for _, event := range events { + for _, link := range event.Payloads { + k := link.Ref.Store + "\x00" + link.Ref.Key + if seen[k] { + continue + } + seen[k] = true + source := w.sources[link.Ref.Store] + if source == nil { + return Receipt{}, errors.New("trail/archive: source payload store is not configured") + } + reader, err := source.Open(ctx, link.Ref) + if err != nil { + return Receipt{}, err + } + logical := io.Reader(reader) + var gz *gzip.Reader + if link.Ref.Compression == payload.CompressionGZIP { + gz, err = gzip.NewReader(reader) + if err != nil { + _ = reader.Close() + return Receipt{}, err + } + logical = gz + } + archived, err := w.archive.Put(ctx, logical, payload.PutOptions{ContentType: link.Ref.ContentType, Compression: link.Ref.Compression, RetainUntil: link.Ref.RetainUntil, RetentionClass: link.Ref.RetentionClass}) + if gz != nil { + _ = gz.Close() + } + _ = reader.Close() + if err != nil { + return Receipt{}, err + } + if archived.SHA256 != link.Ref.SHA256 || archived.Size != link.Ref.Size { + return Receipt{}, errors.New("trail/archive: payload verification failed") + } + manifest.Payloads = append(manifest.Payloads, PayloadEntry{Source: link.Ref, Archive: archived}) + } + } + data, err := json.Marshal(manifest) + if err != nil { + return Receipt{}, err + } + manifestRef, err := w.archive.Put(ctx, bytes.NewReader(data), payload.PutOptions{ContentType: "application/vnd.trail.archive-manifest+json"}) + if err != nil { + return Receipt{}, err + } + opened, err := w.archive.Open(ctx, manifestRef) + if err != nil { + return Receipt{}, err + } + verified, err := io.ReadAll(io.LimitReader(opened, int64(len(data))+1)) + _ = opened.Close() + if err != nil || !bytes.Equal(verified, data) { + return Receipt{}, errors.New("trail/archive: manifest verification failed") + } + return Receipt{Manifest: manifest, ManifestRef: manifestRef}, nil +} + +// ReadManifest reads and verifies a manifest through a previously trusted +// reference. Callers should obtain the reference from the retention catalog, +// never from untrusted request input. +func ReadManifest(ctx context.Context, store payload.Store, ref payload.Ref) (Manifest, error) { + if store == nil { + return Manifest{}, errors.New("trail/archive: archive store is required") + } + r, err := store.Open(ctx, ref) + if err != nil { + return Manifest{}, err + } + defer r.Close() + data, err := io.ReadAll(io.LimitReader(r, 16<<20)) + if err != nil { + return Manifest{}, err + } + if sum := sha256.Sum256(data); sum != ref.SHA256 { + return Manifest{}, errors.New("trail/archive: manifest checksum verification failed") + } + var manifest Manifest + if err = json.Unmarshal(data, &manifest); err != nil { + return Manifest{}, err + } + if manifest.Version != Version || manifest.ID == "" || manifest.EventCount < 1 || len(manifest.EventIDs) != manifest.EventCount { + return Manifest{}, errors.New("trail/archive: invalid manifest") + } + return manifest, nil +} + +// ReadEvents returns at most limit events beginning at offset. It never loads +// an unbounded archive into memory. +func ReadEvents(ctx context.Context, store payload.Store, manifest Manifest, offset, limit int) ([]storage.EventRecord, bool, error) { + if store == nil || offset < 0 || limit < 1 || limit > 10_000 { + return nil, false, errors.New("trail/archive: invalid bounded read") + } + r, err := store.Open(ctx, manifest.Events) + if err != nil { + return nil, false, err + } + defer r.Close() + logical := io.Reader(r) + var gz *gzip.Reader + if manifest.Events.Compression == payload.CompressionGZIP { + gz, err = gzip.NewReader(r) + if err != nil { + return nil, false, err + } + defer gz.Close() + logical = gz + } + hash := sha256.New() + scanner := bufio.NewScanner(io.TeeReader(logical, hash)) + scanner.Buffer(make([]byte, 64<<10), 16<<20) + index := 0 + result := make([]storage.EventRecord, 0, limit) + for scanner.Scan() { + if err = ctx.Err(); err != nil { + return nil, false, err + } + if index >= offset && len(result) < limit { + var event storage.EventRecord + if err = json.Unmarshal(scanner.Bytes(), &event); err != nil { + return nil, false, err + } + result = append(result, event) + } + index++ + } + if err = scanner.Err(); err != nil { + return nil, false, err + } + if index != manifest.EventCount { + return nil, false, errors.New("trail/archive: event count verification failed") + } + var sum [32]byte + copy(sum[:], hash.Sum(nil)) + if sum != manifest.Events.SHA256 { + return nil, false, errors.New("trail/archive: event checksum verification failed") + } + return result, offset+len(result) >= manifest.EventCount, nil +} + +// Restore copies referenced payloads first and then idempotently merges a +// bounded event chunk into the active event store. +func Restore(ctx context.Context, archiveStore payload.Store, activeStores map[string]payload.Store, target RestoreStore, receipt Receipt, offset, limit int) (RestoreResult, error) { + result := RestoreResult{ArchiveID: receipt.Manifest.ID} + if target == nil { + return result, errors.New("trail/archive: restore store is required") + } + manifest, err := ReadManifest(ctx, archiveStore, receipt.ManifestRef) + if err != nil { + return result, err + } + if manifest.ID != receipt.Manifest.ID { + return result, errors.New("trail/archive: manifest identity mismatch") + } + events, complete, err := ReadEvents(ctx, archiveStore, manifest, offset, limit) + if err != nil { + return result, err + } + needed := map[string]bool{} + for _, event := range events { + for _, link := range event.Payloads { + needed[link.Ref.Store+"\x00"+link.Ref.Key] = true + } + } + for _, entry := range manifest.Payloads { + if !needed[entry.Source.Store+"\x00"+entry.Source.Key] { + continue + } + destination := activeStores[entry.Source.Store] + if destination == nil { + return result, errors.New("trail/archive: destination payload store is not configured") + } + reader, er := archiveStore.Open(ctx, entry.Archive) + if er != nil { + return result, er + } + logical := io.Reader(reader) + var gz *gzip.Reader + if entry.Archive.Compression == payload.CompressionGZIP { + gz, er = gzip.NewReader(reader) + if er != nil { + _ = reader.Close() + return result, er + } + logical = gz + } + restored, er := destination.Put(ctx, logical, payload.PutOptions{ContentType: entry.Source.ContentType, Compression: entry.Source.Compression, RetainUntil: entry.Source.RetainUntil, RetentionClass: entry.Source.RetentionClass}) + if gz != nil { + _ = gz.Close() + } + _ = reader.Close() + if er != nil { + return result, er + } + if restored.SHA256 != entry.Source.SHA256 || restored.Size != entry.Source.Size { + return result, errors.New("trail/archive: restored payload verification failed") + } + } + result.Restored, err = target.RestoreArchiveEvents(ctx, manifest.ID, events) + if err != nil { + return result, err + } + result.Complete = complete + if !complete { + result.Next = offset + len(events) + } + return result, nil +} diff --git a/archive/archive_test.go b/archive/archive_test.go new file mode 100644 index 0000000..5f93772 --- /dev/null +++ b/archive/archive_test.go @@ -0,0 +1,111 @@ +package archive + +import ( + "context" + "errors" + "io" + "strings" + "testing" + "time" + + "github.com/vestavision/trail" + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/payload/filesystem" + "github.com/vestavision/trail/storage" +) + +type restoreStore struct { + events map[trail.EventID]storage.EventRecord +} + +func (s *restoreStore) RestoreArchiveEvents(_ context.Context, _ string, events []storage.EventRecord) (int, error) { + if s.events == nil { + s.events = map[trail.EventID]storage.EventRecord{} + } + for _, event := range events { + if _, exists := s.events[event.ID]; !exists { + s.events[event.ID] = event + } + } + return len(events), nil +} + +func TestBundleRoundTripAndIdempotentRestore(t *testing.T) { + ctx := context.Background() + source, err := filesystem.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + archiveStore, err := filesystem.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + ref, err := source.Put(ctx, strings.NewReader("large response"), payload.PutOptions{ContentType: "application/xml", Compression: payload.CompressionGZIP, RetentionClass: "http"}) + if err != nil { + t.Fatal(err) + } + id1, _ := trail.ParseEventID("00000000000000000000000001") + id2, _ := trail.ParseEventID("00000000000000000000000002") + events := []storage.EventRecord{ + {ID: id1, Timestamp: time.Unix(1, 0).UTC(), Kind: "provider.http", Payloads: []storage.PayloadLink{{Role: "response", Ref: ref}}}, + {ID: id2, Timestamp: time.Unix(2, 0).UTC(), Kind: "payment.completed"}, + } + w, err := NewWriter(archiveStore, map[string]payload.Store{"filesystem": source}) + if err != nil { + t.Fatal(err) + } + receipt, err := w.Write(ctx, events) + if err != nil { + t.Fatal(err) + } + manifest, err := ReadManifest(ctx, archiveStore, receipt.ManifestRef) + if err != nil { + t.Fatal(err) + } + if manifest.EventCount != 2 || len(manifest.Payloads) != 1 { + t.Fatalf("manifest=%+v", manifest) + } + page, complete, err := ReadEvents(ctx, archiveStore, manifest, 0, 1) + if err != nil || complete || len(page) != 1 || page[0].ID != id1 { + t.Fatalf("page=%+v complete=%v err=%v", page, complete, err) + } + restoredPayloads, _ := filesystem.New(t.TempDir()) + target := &restoreStore{} + r1, err := Restore(ctx, archiveStore, map[string]payload.Store{"filesystem": restoredPayloads}, target, receipt, 0, 1) + if err != nil || r1.Complete || r1.Next != 1 { + t.Fatalf("first restore=%+v err=%v", r1, err) + } + r2, err := Restore(ctx, archiveStore, map[string]payload.Store{"filesystem": restoredPayloads}, target, receipt, r1.Next, 10) + if err != nil || !r2.Complete || len(target.events) != 2 { + t.Fatalf("second restore=%+v events=%d err=%v", r2, len(target.events), err) + } + if _, err = Restore(ctx, archiveStore, map[string]payload.Store{"filesystem": restoredPayloads}, target, receipt, 0, 10); err != nil || len(target.events) != 2 { + t.Fatalf("idempotent restore events=%d err=%v", len(target.events), err) + } + opened, err := restoredPayloads.Open(ctx, ref) + if err != nil { + t.Fatal(err) + } + _ = opened.Close() +} + +type failingStore struct{ err error } + +func (s failingStore) Put(context.Context, io.Reader, payload.PutOptions) (payload.Ref, error) { + return payload.Ref{}, s.err +} +func (s failingStore) Open(context.Context, payload.Ref) (io.ReadCloser, error) { + return nil, s.err +} +func (s failingStore) Delete(context.Context, payload.Ref) error { return s.err } + +func TestArchiveWriteFailureHasNoReceipt(t *testing.T) { + w, err := NewWriter(failingStore{err: errors.New("archive unavailable")}, nil) + if err != nil { + t.Fatal(err) + } + id, _ := trail.ParseEventID("00000000000000000000000001") + if receipt, err := w.Write(context.Background(), []storage.EventRecord{{ID: id}}); err == nil || receipt.Manifest.ID != "" { + t.Fatalf("receipt=%+v err=%v", receipt, err) + } +} diff --git a/cmd/trail-explorer-api/main.go b/cmd/trail-explorer-api/main.go index f4eb375..a1471bc 100644 --- a/cmd/trail-explorer-api/main.go +++ b/cmd/trail-explorer-api/main.go @@ -12,6 +12,7 @@ import ( "syscall" "time" + "github.com/vestavision/trail/archive" "github.com/vestavision/trail/explorerapi" "github.com/vestavision/trail/payload" "github.com/vestavision/trail/payload/filesystem" @@ -55,7 +56,14 @@ func run(ctx context.Context) error { } payloads["s3"] = s3 } - handler, err := explorerapi.New(explorerapi.Config{Store: store, PayloadStores: payloads, MaxPayloadBytes: int64(envInt("TRAIL_EXPLORER_MAX_PAYLOAD_BYTES", 1<<20)), AllowedOrigins: split(os.Getenv("TRAIL_EXPLORER_ALLOWED_ORIGINS"))}) + var archiveStore payload.Store + if os.Getenv("TRAIL_ARCHIVE_STORE") != "" { + archiveStore, err = openArchiveStore() + if err != nil { + return err + } + } + handler, err := explorerapi.New(explorerapi.Config{Store: store, PayloadStores: payloads, MaxPayloadBytes: int64(envInt("TRAIL_EXPLORER_MAX_PAYLOAD_BYTES", 1<<20)), AllowedOrigins: split(os.Getenv("TRAIL_EXPLORER_ALLOWED_ORIGINS")), ArchiveCatalog: store, ArchiveStore: archiveStore, ArchiveRestore: store, EnableRestore: envBool("TRAIL_EXPLORER_ENABLE_ARCHIVE_RESTORE", false)}) if err != nil { return err } @@ -72,7 +80,12 @@ func run(ctx context.Context) error { } return err } -func openStore(ctx context.Context) (storage.ExplorerStore, error) { +func openStore(ctx context.Context) (interface { + storage.ExplorerStore + archive.Catalog + archive.RestoreStore + Close() error +}, error) { switch env("TRAIL_STORE", "clickhouse") { case "clickhouse": s, e := trailch.Open(trailch.Config{Addresses: []string{env("TRAIL_CLICKHOUSE_ADDR", "127.0.0.1:9000")}, Database: env("TRAIL_CLICKHOUSE_DATABASE", "default"), Username: env("TRAIL_CLICKHOUSE_USERNAME", "default"), Password: os.Getenv("TRAIL_CLICKHOUSE_PASSWORD")}) @@ -98,6 +111,25 @@ func openStore(ctx context.Context) (storage.ExplorerStore, error) { return nil, fmt.Errorf("unknown TRAIL_STORE") } } + +func openArchiveStore() (payload.Store, error) { + switch env("TRAIL_ARCHIVE_STORE", "filesystem") { + case "filesystem": + root := os.Getenv("TRAIL_ARCHIVE_FILESYSTEM_ROOT") + if root == "" { + return nil, errors.New("TRAIL_ARCHIVE_FILESYSTEM_ROOT is required") + } + return filesystem.New(root) + case "s3": + endpoint := os.Getenv("TRAIL_ARCHIVE_S3_ENDPOINT") + if endpoint == "" { + return nil, errors.New("TRAIL_ARCHIVE_S3_ENDPOINT is required") + } + return trails3.Open(trails3.Config{Endpoint: endpoint, AccessKey: os.Getenv("TRAIL_ARCHIVE_S3_ACCESS_KEY"), SecretKey: os.Getenv("TRAIL_ARCHIVE_S3_SECRET_KEY"), Bucket: env("TRAIL_ARCHIVE_S3_BUCKET", "trail-archives"), Region: os.Getenv("TRAIL_ARCHIVE_S3_REGION"), Prefix: env("TRAIL_ARCHIVE_S3_PREFIX", "archives"), Secure: envBool("TRAIL_ARCHIVE_S3_SECURE", false), PathStyle: envBool("TRAIL_ARCHIVE_S3_PATH_STYLE", true), Name: "archive-s3"}) + default: + return nil, errors.New("unknown TRAIL_ARCHIVE_STORE") + } +} func env(k, v string) string { if x := os.Getenv(k); x != "" { return x diff --git a/cmd/trail-retention/main.go b/cmd/trail-retention/main.go new file mode 100644 index 0000000..4bfc9c0 --- /dev/null +++ b/cmd/trail-retention/main.go @@ -0,0 +1,362 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "os" + "os/signal" + "path/filepath" + "strconv" + "strings" + "syscall" + "time" + + "github.com/vestavision/trail" + "github.com/vestavision/trail/archive" + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/payload/filesystem" + trails3 "github.com/vestavision/trail/payload/s3" + "github.com/vestavision/trail/retention" + trailch "github.com/vestavision/trail/storage/clickhouse" + trailpg "github.com/vestavision/trail/storage/postgres" +) + +type duration time.Duration + +func (d *duration) UnmarshalJSON(data []byte) error { + var v string + if err := json.Unmarshal(data, &v); err != nil { + return errors.New("durations must be strings such as \"168h\"") + } + x, err := time.ParseDuration(v) + *d = duration(x) + return err +} + +type matchFile struct { + Kind string `json:"kind,omitempty"` + Service string `json:"service,omitempty"` + Environment string `json:"environment,omitempty"` + Level string `json:"level,omitempty"` + FlowStatus string `json:"flow_status,omitempty"` + ExecutionStatus string `json:"execution_status,omitempty"` + HTTPStatusClass int `json:"http_status_class,omitempty"` + HTTPSuccess *bool `json:"http_success,omitempty"` +} +type eventPolicyFile struct { + Name string `json:"name"` + Match matchFile `json:"match"` + TTL duration `json:"ttl"` +} +type payloadMatchFile struct { + matchFile + Role string `json:"role,omitempty"` + ContentType string `json:"content_type,omitempty"` + RetentionClass string `json:"retention_class,omitempty"` +} +type payloadPolicyFile struct { + Name string `json:"name"` + Match payloadMatchFile `json:"match"` + TTL duration `json:"ttl"` +} +type configFile struct { + EventPolicies []eventPolicyFile `json:"event_policies"` + EventClasses map[string]duration `json:"event_classes"` + DefaultEventTTL duration `json:"default_event_ttl"` + PayloadPolicies []payloadPolicyFile `json:"payload_policies"` + PayloadClasses map[string]duration `json:"payload_classes"` + DefaultPayloadTTL duration `json:"default_payload_ttl"` + EventBatchSize int `json:"event_batch_size"` + PayloadBatchSize int `json:"payload_batch_size"` + PayloadConcurrency int `json:"payload_concurrency"` + MaxEvents int `json:"max_events"` + MaxPayloads int `json:"max_payloads"` + MaxRuntime duration `json:"max_runtime"` + PayloadGracePeriod duration `json:"payload_grace_period"` +} + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, os.Args[1:]); err != nil && !errors.Is(err, context.Canceled) { + fmt.Fprintln(os.Stderr, "trail-retention:", err) + os.Exit(1) + } +} +func run(ctx context.Context, args []string) error { + if len(args) == 0 { + return errors.New("usage: trail-retention ") + } + if args[0] != "run" { + return runArchiveCommand(ctx, args) + } + fs := flag.NewFlagSet("run", flag.ContinueOnError) + path := fs.String("config", env("TRAIL_RETENTION_CONFIG", ""), "retention policy JSON") + dry := fs.Bool("dry-run", false, "report without deleting") + maxEvents := fs.Int("max-events", 0, "override maximum events examined") + maxPayloads := fs.Int("max-payloads", 0, "override maximum payload candidates") + maxRuntime := fs.Duration("max-runtime", 0, "override maximum run duration") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *path == "" { + return errors.New("--config or TRAIL_RETENTION_CONFIG is required") + } + cfg, err := loadConfig(*path) + if err != nil { + return err + } + if *maxEvents > 0 { + cfg.MaxEvents = *maxEvents + } + if *maxPayloads > 0 { + cfg.MaxPayloads = *maxPayloads + } + if *maxRuntime > 0 { + cfg.MaxRuntime = *maxRuntime + } + store, err := openStore(ctx) + if err != nil { + return err + } + defer store.Close() + payloads, err := openPayloads() + if err != nil { + return err + } + archiveStore, err := openArchiveStore() + if err != nil { + return err + } + archiver, err := archive.NewWriter(archiveStore, payloads) + if err != nil { + return err + } + cfg.Archiver = archiver + engine, err := retention.New(cfg, store, payloads) + if err != nil { + return err + } + result, err := engine.Run(ctx, retention.RunOptions{DryRun: *dry}) + _ = json.NewEncoder(os.Stdout).Encode(result) + return err +} + +func runArchiveCommand(ctx context.Context, args []string) error { + fs := flag.NewFlagSet(args[0], flag.ContinueOnError) + id := fs.String("archive-id", "", "catalog archive ID") + offset := fs.Int("offset", 0, "event offset") + limit := fs.Int("limit", 100, "bounded event limit (1-10000)") + if err := fs.Parse(args[1:]); err != nil { + return err + } + store, err := openStore(ctx) + if err != nil { + return err + } + defer store.Close() + switch args[0] { + case "archives": + page, err := store.ListArchives(ctx, *offset, min(*limit, 500)) + if err == nil { + err = json.NewEncoder(os.Stdout).Encode(page) + } + return err + case "inspect": + if *id == "" { + return errors.New("--archive-id is required") + } + receipt, err := store.GetArchive(ctx, *id) + if err == nil { + err = json.NewEncoder(os.Stdout).Encode(receipt) + } + return err + case "restore": + if *id == "" { + return errors.New("--archive-id is required") + } + archiveStore, err := openArchiveStore() + if err != nil { + return err + } + payloads, err := openPayloads() + if err != nil { + return err + } + receipt, err := store.GetArchive(ctx, *id) + if err != nil { + return err + } + result, err := archive.Restore(ctx, archiveStore, payloads, store, receipt, *offset, *limit) + _ = json.NewEncoder(os.Stdout).Encode(result) + return err + default: + return errors.New("usage: trail-retention ") + } +} + +func openArchiveStore() (payload.Store, error) { + switch env("TRAIL_ARCHIVE_STORE", "filesystem") { + case "filesystem": + root := os.Getenv("TRAIL_ARCHIVE_FILESYSTEM_ROOT") + if root == "" { + return nil, errors.New("TRAIL_ARCHIVE_FILESYSTEM_ROOT is required") + } + archiveRoot, err := filepath.Abs(root) + if err != nil { + return nil, err + } + if active := os.Getenv("TRAIL_PAYLOAD_FILESYSTEM_ROOT"); active != "" { + activeRoot, err := filepath.Abs(active) + if err != nil { + return nil, err + } + if archiveRoot == activeRoot { + return nil, errors.New("archive and active filesystem roots must be different") + } + } + return filesystem.New(archiveRoot) + case "s3": + endpoint := os.Getenv("TRAIL_ARCHIVE_S3_ENDPOINT") + if endpoint == "" { + return nil, errors.New("TRAIL_ARCHIVE_S3_ENDPOINT is required") + } + archiveBucket := env("TRAIL_ARCHIVE_S3_BUCKET", "trail-archives") + archivePrefix := env("TRAIL_ARCHIVE_S3_PREFIX", "archives") + if endpoint == os.Getenv("TRAIL_PAYLOAD_S3_ENDPOINT") && archiveBucket == env("TRAIL_PAYLOAD_S3_BUCKET", "trail-payloads") && prefixesOverlap(archivePrefix, os.Getenv("TRAIL_PAYLOAD_S3_PREFIX")) { + return nil, errors.New("archive and active S3 prefixes must not overlap") + } + return trails3.Open(trails3.Config{Endpoint: endpoint, AccessKey: os.Getenv("TRAIL_ARCHIVE_S3_ACCESS_KEY"), SecretKey: os.Getenv("TRAIL_ARCHIVE_S3_SECRET_KEY"), Bucket: archiveBucket, Region: os.Getenv("TRAIL_ARCHIVE_S3_REGION"), Prefix: archivePrefix, Secure: envBool("TRAIL_ARCHIVE_S3_SECURE", false), PathStyle: envBool("TRAIL_ARCHIVE_S3_PATH_STYLE", true), Name: "archive-s3"}) + default: + return nil, errors.New("unknown TRAIL_ARCHIVE_STORE") + } +} +func loadConfig(path string) (retention.Config, error) { + data, err := os.ReadFile(path) + if err != nil { + return retention.Config{}, err + } + var f configFile + if err = json.Unmarshal(data, &f); err != nil { + return retention.Config{}, err + } + c := retention.Config{DefaultEventTTL: time.Duration(f.DefaultEventTTL), DefaultPayloadTTL: time.Duration(f.DefaultPayloadTTL), EventBatchSize: f.EventBatchSize, PayloadBatchSize: f.PayloadBatchSize, PayloadConcurrency: f.PayloadConcurrency, MaxEvents: f.MaxEvents, MaxPayloads: f.MaxPayloads, MaxRuntime: time.Duration(f.MaxRuntime), PayloadGracePeriod: time.Duration(f.PayloadGracePeriod), EventClasses: map[string]time.Duration{}, PayloadClasses: map[string]time.Duration{}} + for k, v := range f.EventClasses { + c.EventClasses[k] = time.Duration(v) + } + for k, v := range f.PayloadClasses { + c.PayloadClasses[k] = time.Duration(v) + } + for _, p := range f.EventPolicies { + m, err := eventMatch(p.Match) + if err != nil { + return c, err + } + c.EventPolicies = append(c.EventPolicies, retention.EventPolicy{Name: p.Name, Match: m, TTL: time.Duration(p.TTL)}) + } + for _, p := range f.PayloadPolicies { + m, err := eventMatch(p.Match.matchFile) + if err != nil { + return c, err + } + c.PayloadPolicies = append(c.PayloadPolicies, retention.PayloadPolicy{Name: p.Name, TTL: time.Duration(p.TTL), Match: retention.PayloadMatch{EventMatch: m, Role: p.Match.Role, ContentType: p.Match.ContentType, RetentionClass: p.Match.RetentionClass}}) + } + if len(c.EventPolicies) == 0 && len(c.EventClasses) == 0 && c.DefaultEventTTL == 0 { + return c, errors.New("at least one event retention policy, class, or default TTL is required") + } + return c, nil +} +func eventMatch(f matchFile) (retention.EventMatch, error) { + m := retention.EventMatch{Kind: f.Kind, Service: f.Service, Environment: f.Environment, FlowStatus: f.FlowStatus, ExecutionStatus: f.ExecutionStatus, HTTPStatusClass: f.HTTPStatusClass, HTTPSuccess: f.HTTPSuccess} + if f.Level != "" { + var l trail.Level + switch f.Level { + case "debug": + l = trail.LevelDebug + case "info": + l = trail.LevelInfo + case "warn": + l = trail.LevelWarn + case "error": + l = trail.LevelError + default: + return m, fmt.Errorf("invalid level %q", f.Level) + } + m.Level = &l + } + return m, nil +} +func openStore(ctx context.Context) (interface { + retention.Store + archive.Catalog + archive.RestoreStore + Close() error +}, error) { + switch env("TRAIL_STORE", "clickhouse") { + case "clickhouse": + s, e := trailch.Open(trailch.Config{Addresses: []string{env("TRAIL_CLICKHOUSE_ADDR", "127.0.0.1:9000")}, Database: env("TRAIL_CLICKHOUSE_DATABASE", "default"), Username: env("TRAIL_CLICKHOUSE_USERNAME", "default"), Password: os.Getenv("TRAIL_CLICKHOUSE_PASSWORD")}) + if e == nil { + e = s.Migrate(ctx) + } + return s, e + case "postgres": + s, e := trailpg.Open(ctx, trailpg.Config{URL: env("TRAIL_POSTGRES_URL", "postgres://trail:trail@127.0.0.1:5432/trail?sslmode=disable"), MaxConnections: int32(envInt("TRAIL_POSTGRES_MAX_CONNECTIONS", 4))}) + if e == nil { + e = s.Migrate(ctx) + } + return s, e + default: + return nil, errors.New("unknown TRAIL_STORE") + } +} +func openPayloads() (map[string]payload.Store, error) { + out := map[string]payload.Store{} + if root := os.Getenv("TRAIL_PAYLOAD_FILESYSTEM_ROOT"); root != "" { + s, e := filesystem.New(root) + if e != nil { + return nil, e + } + out["filesystem"] = s + } + if endpoint := os.Getenv("TRAIL_PAYLOAD_S3_ENDPOINT"); endpoint != "" { + s, e := trails3.Open(trails3.Config{Endpoint: endpoint, AccessKey: os.Getenv("TRAIL_PAYLOAD_S3_ACCESS_KEY"), SecretKey: os.Getenv("TRAIL_PAYLOAD_S3_SECRET_KEY"), Bucket: env("TRAIL_PAYLOAD_S3_BUCKET", "trail-payloads"), Region: os.Getenv("TRAIL_PAYLOAD_S3_REGION"), Prefix: os.Getenv("TRAIL_PAYLOAD_S3_PREFIX"), Secure: envBool("TRAIL_PAYLOAD_S3_SECURE", false), PathStyle: envBool("TRAIL_PAYLOAD_S3_PATH_STYLE", true), Name: "s3"}) + if e != nil { + return nil, e + } + out["s3"] = s + } + return out, nil +} + +func prefixesOverlap(a, b string) bool { + a = strings.Trim(strings.TrimSpace(a), "/") + b = strings.Trim(strings.TrimSpace(b), "/") + return a == b || a == "" || b == "" || strings.HasPrefix(a+"/", b+"/") || strings.HasPrefix(b+"/", a+"/") +} +func env(k, v string) string { + if x := os.Getenv(k); x != "" { + return x + } + return v +} +func envInt(k string, v int) int { + x, e := strconv.Atoi(os.Getenv(k)) + if e != nil || x <= 0 { + return v + } + return x +} +func envBool(k string, v bool) bool { + x := os.Getenv(k) + if x == "" { + return v + } + b, e := strconv.ParseBool(x) + if e != nil { + return v + } + return b +} diff --git a/cmd/trail-retention/main_test.go b/cmd/trail-retention/main_test.go new file mode 100644 index 0000000..b512234 --- /dev/null +++ b/cmd/trail-retention/main_test.go @@ -0,0 +1,36 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestLoadConfig(t *testing.T) { + path := filepath.Join(t.TempDir(), "retention.json") + data := []byte(`{"event_policies":[{"name":"http","match":{"kind":"provider.http","level":"warn","http_status_class":5},"ttl":"24h"}],"payload_policies":[{"name":"xml","match":{"role":"response","content_type":"application/xml"},"ttl":"48h"}],"payload_grace_period":"2h","payload_concurrency":2}`) + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + cfg, err := loadConfig(path) + if err != nil { + t.Fatal(err) + } + if len(cfg.EventPolicies) != 1 || cfg.EventPolicies[0].Match.Kind != "provider.http" || cfg.EventPolicies[0].TTL != 24*time.Hour { + t.Fatalf("event config=%+v", cfg.EventPolicies) + } + if len(cfg.PayloadPolicies) != 1 || cfg.PayloadPolicies[0].Match.Role != "response" || cfg.PayloadGracePeriod != 2*time.Hour || cfg.PayloadConcurrency != 2 { + t.Fatalf("payload config=%+v", cfg) + } +} + +func TestLoadConfigRequiresEventPolicy(t *testing.T) { + path := filepath.Join(t.TempDir(), "empty.json") + if err := os.WriteFile(path, []byte(`{"payload_classes":{"short":"24h"}}`), 0o600); err != nil { + t.Fatal(err) + } + if _, err := loadConfig(path); err == nil { + t.Fatal("expected safe configuration error") + } +} diff --git a/convention/convention.go b/convention/convention.go index 93ce80f..d56de43 100644 --- a/convention/convention.go +++ b/convention/convention.go @@ -9,6 +9,7 @@ const ( FieldFlowStatus = "trail.flow.status" FieldExecutionStatus = "trail.execution.status" FieldExecutionKind = "trail.execution.kind" + FieldRetentionClass = "trail.retention.class" ) type Status string @@ -32,6 +33,12 @@ func ExecutionKind(kind string) trail.Option { return trail.String(FieldExecutionKind, kind) } +// RetentionClass assigns an optional lifecycle class to an event. It has no +// effect unless an external retention engine is configured for that class. +func RetentionClass(class string) trail.Option { + return trail.String(FieldRetentionClass, class) +} + func IsCanonicalStatus(status string) bool { switch Status(status) { case StatusRunning, StatusSucceeded, StatusFailed, StatusCancelled: diff --git a/explorerapi/server.go b/explorerapi/server.go index 24a7503..64df2e8 100644 --- a/explorerapi/server.go +++ b/explorerapi/server.go @@ -13,6 +13,7 @@ import ( "time" "github.com/vestavision/trail" + "github.com/vestavision/trail/archive" "github.com/vestavision/trail/payload" "github.com/vestavision/trail/storage" ) @@ -22,6 +23,10 @@ type Config struct { PayloadStores map[string]payload.Store MaxPayloadBytes int64 AllowedOrigins []string + ArchiveCatalog archive.Catalog + ArchiveStore payload.Store + ArchiveRestore archive.RestoreStore + EnableRestore bool } type Server struct { @@ -53,10 +58,120 @@ func New(cfg Config) (*Server, error) { mux.HandleFunc("GET /api/v1/events", s.events) mux.HandleFunc("GET /api/v1/events/{id}", s.event) mux.HandleFunc("GET /api/v1/events/{event}/payloads/{role}", s.eventPayload) + mux.HandleFunc("GET /api/v1/archives", s.archives) + mux.HandleFunc("GET /api/v1/archives/{id}", s.archiveDetail) + mux.HandleFunc("GET /api/v1/archives/{id}/events", s.archiveEvents) + mux.HandleFunc("GET /api/v1/archives/{id}/payloads/{index}", s.archivePayload) + mux.HandleFunc("POST /api/v1/archives/{id}/restore", s.restoreArchive) s.handler = recoverMiddleware(cors(cfg.AllowedOrigins, mux)) return s, nil } +func (s *Server) archives(w http.ResponseWriter, r *http.Request) { + if s.cfg.ArchiveCatalog == nil { + respondError(w, http.StatusNotImplemented, "capability_unavailable", "archive catalog is not configured") + return + } + offset, _ := strconv.Atoi(r.URL.Query().Get("offset")) + limit := 100 + if raw := r.URL.Query().Get("limit"); raw != "" { + limit, _ = strconv.Atoi(raw) + } + if offset < 0 || limit < 1 || limit > 500 { + bad(w, errors.New("offset and limit are invalid")) + return + } + v, err := s.cfg.ArchiveCatalog.ListArchives(r.Context(), offset, limit) + respond(w, v, err) +} + +func (s *Server) archiveDetail(w http.ResponseWriter, r *http.Request) { + if s.cfg.ArchiveCatalog == nil { + respondError(w, http.StatusNotImplemented, "capability_unavailable", "archive catalog is not configured") + return + } + v, err := s.cfg.ArchiveCatalog.GetArchive(r.Context(), r.PathValue("id")) + respond(w, v, err) +} + +func (s *Server) archiveEvents(w http.ResponseWriter, r *http.Request) { + receipt, offset, limit, ok := s.archiveRequest(w, r) + if !ok { + return + } + events, complete, err := archive.ReadEvents(r.Context(), s.cfg.ArchiveStore, receipt.Manifest, offset, limit) + respond(w, map[string]any{"items": events, "next_offset": offset + len(events), "complete": complete}, err) +} + +func (s *Server) archivePayload(w http.ResponseWriter, r *http.Request) { + if s.cfg.ArchiveCatalog == nil || s.cfg.ArchiveStore == nil { + respondError(w, http.StatusNotImplemented, "capability_unavailable", "archive access is not configured") + return + } + receipt, err := s.cfg.ArchiveCatalog.GetArchive(r.Context(), r.PathValue("id")) + if err != nil { + respond(w, nil, err) + return + } + index, err := strconv.Atoi(r.PathValue("index")) + if err != nil || index < 0 || index >= len(receipt.Manifest.Payloads) { + bad(w, errors.New("invalid payload index")) + return + } + ref := receipt.Manifest.Payloads[index].Archive + if ref.Size > s.cfg.MaxPayloadBytes { + respondError(w, http.StatusRequestEntityTooLarge, "payload_too_large", "payload exceeds Explorer read limit") + return + } + reader, err := s.cfg.ArchiveStore.Open(r.Context(), ref) + if err != nil { + respond(w, nil, err) + return + } + defer reader.Close() + w.Header().Set("Content-Type", safeContentType(ref.ContentType)) + w.Header().Set("X-Content-Type-Options", "nosniff") + if ref.Compression == payload.CompressionGZIP { + w.Header().Set("Content-Encoding", "gzip") + } + _, _ = io.Copy(w, io.LimitReader(reader, s.cfg.MaxPayloadBytes)) +} + +func (s *Server) restoreArchive(w http.ResponseWriter, r *http.Request) { + if !s.cfg.EnableRestore || s.cfg.ArchiveRestore == nil { + respondError(w, http.StatusNotFound, "not_found", "archive restore is not enabled") + return + } + receipt, offset, limit, ok := s.archiveRequest(w, r) + if !ok { + return + } + v, err := archive.Restore(r.Context(), s.cfg.ArchiveStore, s.cfg.PayloadStores, s.cfg.ArchiveRestore, receipt, offset, limit) + respond(w, v, err) +} + +func (s *Server) archiveRequest(w http.ResponseWriter, r *http.Request) (archive.Receipt, int, int, bool) { + if s.cfg.ArchiveCatalog == nil || s.cfg.ArchiveStore == nil { + respondError(w, http.StatusNotImplemented, "capability_unavailable", "archive access is not configured") + return archive.Receipt{}, 0, 0, false + } + offset, _ := strconv.Atoi(r.URL.Query().Get("offset")) + limit := 100 + if raw := r.URL.Query().Get("limit"); raw != "" { + limit, _ = strconv.Atoi(raw) + } + if offset < 0 || limit < 1 || limit > 10_000 { + bad(w, errors.New("offset and limit are invalid")) + return archive.Receipt{}, 0, 0, false + } + receipt, err := s.cfg.ArchiveCatalog.GetArchive(r.Context(), r.PathValue("id")) + if err != nil { + respond(w, nil, err) + return archive.Receipt{}, 0, 0, false + } + return receipt, offset, limit, true +} + func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.handler.ServeHTTP(w, r) } func (s *Server) overview(w http.ResponseWriter, r *http.Request) { diff --git a/retention.example.json b/retention.example.json new file mode 100644 index 0000000..8ed9a2b --- /dev/null +++ b/retention.example.json @@ -0,0 +1,29 @@ +{ + "event_policies": [ + { + "name": "http-success", + "match": { "kind": "provider.http", "http_success": true }, + "ttl": "168h" + }, + { + "name": "http-error", + "match": { "kind": "provider.http", "http_success": false }, + "ttl": "720h" + }, + { "name": "business", "match": {}, "ttl": "2160h" } + ], + "event_classes": { "audit": "8760h" }, + "payload_classes": { + "http": "168h", + "error": "720h", + "audit": "8760h" + }, + "default_payload_ttl": "168h", + "event_batch_size": 1000, + "payload_batch_size": 100, + "payload_concurrency": 1, + "max_events": 100000, + "max_payloads": 10000, + "max_runtime": "10m", + "payload_grace_period": "24h" +} diff --git a/retention/retention.go b/retention/retention.go new file mode 100644 index 0000000..3b61b57 --- /dev/null +++ b/retention/retention.go @@ -0,0 +1,548 @@ +// Package retention provides an optional, bounded lifecycle engine for stored +// Trail events and payloads. It is deliberately independent from Trail's +// logging runtime and never starts background work. +package retention + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/vestavision/trail" + "github.com/vestavision/trail/archive" + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/storage" +) + +type EventMatch struct { + Kind, Service, Environment string + Level *trail.Level + FlowStatus, ExecutionStatus string + HTTPStatusClass int + HTTPSuccess *bool +} + +type EventPolicy struct { + Name string + Match EventMatch + TTL time.Duration +} + +type PayloadMatch struct { + EventMatch + Role, ContentType, RetentionClass string +} + +type PayloadPolicy struct { + Name string + Match PayloadMatch + TTL time.Duration +} + +type Config struct { + EventPolicies []EventPolicy + EventClasses map[string]time.Duration + DefaultEventTTL time.Duration + PayloadPolicies []PayloadPolicy + PayloadClasses map[string]time.Duration + DefaultPayloadTTL time.Duration + EventBatchSize int + PayloadBatchSize int + PayloadConcurrency int + MaxEvents int + MaxPayloads int + MaxRuntime time.Duration + PayloadGracePeriod time.Duration + Now func() time.Time + Archiver archive.Writer +} + +type Event struct { + ID, Cursor string + FlowID, ExecutionID string + Timestamp time.Time + Kind, Service, Environment string + Level trail.Level + FlowStatus, ExecutionStatus string + RetentionClass string + HTTPStatusCode int + HTTPSuccess bool + Payloads []Payload +} + +type Payload struct { + Role string + Ref payload.Ref +} + +type ScanRequest struct { + Cursor string + Limit int +} +type EventPage struct { + Events []Event + NextCursor string + HasMore bool +} +type DeleteRequest struct { + Events []Event + PayloadEligibleAt map[string]time.Time + ArchiveID string + DryRun bool +} +type DeleteResult struct{ EventsDeleted, PayloadCandidates uint64 } + +type Candidate struct { + Ref payload.Ref + ArchiveID string + EligibleAt time.Time + CandidateAt time.Time + Attempts uint32 + LastError string +} +type CandidatePage struct{ Candidates []Candidate } + +// Store is intentionally separate from storage.IngestStore and +// storage.ExplorerStore. Implementations may use database-native deletion. +type Store interface { + ReconcileRetention(context.Context) error + LoadArchiveEvents(context.Context, []Event) ([]storage.EventRecord, error) + RecordArchive(context.Context, archive.Receipt) error + ScanRetentionEvents(context.Context, ScanRequest) (EventPage, error) + DeleteRetentionEvents(context.Context, DeleteRequest) (DeleteResult, error) + ListPayloadCandidates(context.Context, time.Time, int) (CandidatePage, error) + ReferencedPayloads(context.Context, []payload.Ref) (map[string]bool, error) + CompletePayloadCandidate(context.Context, payload.Ref, string) error +} + +type RunOptions struct{ DryRun bool } +type Result struct { + StartedAt time.Time `json:"started_at"` + FinishedAt time.Time `json:"finished_at"` + Duration time.Duration `json:"duration_ns"` + EventsScanned uint64 `json:"events_scanned"` + EventsMatched uint64 `json:"events_matched"` + EventsEligible uint64 `json:"events_eligible"` + EventsDeleted uint64 `json:"events_deleted"` + PayloadCandidates uint64 `json:"payload_candidates"` + PayloadsReferenced uint64 `json:"payloads_referenced"` + PayloadsOrphaned uint64 `json:"payloads_orphaned"` + PayloadsUnknown uint64 `json:"payloads_unknown"` + PayloadsDeleted uint64 `json:"payloads_deleted"` + PayloadBytesReclaimed uint64 `json:"payload_bytes_reclaimed"` + Errors uint64 `json:"errors"` + ArchivesCreated uint64 `json:"archives_created"` + EventsArchived uint64 `json:"events_archived"` + PayloadsArchived uint64 `json:"payloads_archived"` + ArchiveBytesWritten uint64 `json:"archive_bytes_written"` + ArchiveFailures uint64 `json:"archive_failures"` +} + +type Engine struct { + cfg Config + store Store + payloads map[string]payload.Store +} + +func New(cfg Config, store Store, payloads map[string]payload.Store) (*Engine, error) { + if store == nil { + return nil, errors.New("trail/retention: store is required") + } + if cfg.Archiver == nil { + return nil, errors.New("trail/retention: archiver is required") + } + if cfg.EventBatchSize <= 0 { + cfg.EventBatchSize = 1000 + } + if cfg.PayloadBatchSize <= 0 { + cfg.PayloadBatchSize = 100 + } + if cfg.PayloadConcurrency <= 0 { + cfg.PayloadConcurrency = 1 + } + if cfg.PayloadConcurrency > 32 { + return nil, errors.New("trail/retention: payload concurrency exceeds 32") + } + if cfg.MaxEvents <= 0 { + cfg.MaxEvents = 100000 + } + if cfg.MaxPayloads <= 0 { + cfg.MaxPayloads = 10000 + } + if cfg.MaxRuntime <= 0 { + cfg.MaxRuntime = 10 * time.Minute + } + if cfg.PayloadGracePeriod <= 0 { + cfg.PayloadGracePeriod = 24 * time.Hour + } + if cfg.Now == nil { + cfg.Now = func() time.Time { return time.Now().UTC() } + } + if cfg.EventBatchSize > 100000 || cfg.PayloadBatchSize > 10000 || cfg.MaxEvents > 10000000 || cfg.MaxPayloads > 1000000 || cfg.MaxRuntime > 24*time.Hour { + return nil, errors.New("trail/retention: configured work bound exceeds safety limit") + } + if err := validate(cfg); err != nil { + return nil, err + } + return &Engine{cfg: cfg, store: store, payloads: payloads}, nil +} + +func validate(c Config) error { + for _, p := range c.EventPolicies { + if p.TTL <= 0 { + return fmt.Errorf("trail/retention: event policy %q needs positive TTL", p.Name) + } + if p.Match.HTTPStatusClass < 0 || p.Match.HTTPStatusClass > 5 { + return fmt.Errorf("trail/retention: event policy %q has invalid HTTP status class", p.Name) + } + } + for _, p := range c.PayloadPolicies { + if p.TTL <= 0 { + return fmt.Errorf("trail/retention: payload policy %q needs positive TTL", p.Name) + } + if p.Match.HTTPStatusClass < 0 || p.Match.HTTPStatusClass > 5 { + return fmt.Errorf("trail/retention: payload policy %q has invalid HTTP status class", p.Name) + } + } + for n, ttl := range c.EventClasses { + if strings.TrimSpace(n) == "" || ttl <= 0 { + return errors.New("trail/retention: invalid event class") + } + } + for n, ttl := range c.PayloadClasses { + if strings.TrimSpace(n) == "" || ttl <= 0 { + return errors.New("trail/retention: invalid payload class") + } + } + if c.DefaultEventTTL < 0 || c.DefaultPayloadTTL < 0 { + return errors.New("trail/retention: default TTL cannot be negative") + } + return nil +} + +func (e *Engine) Run(ctx context.Context, opt RunOptions) (Result, error) { + now := e.cfg.Now().UTC() + result := Result{StartedAt: now} + ctx, cancel := context.WithTimeout(ctx, e.cfg.MaxRuntime) + defer cancel() + if !opt.DryRun { + if err := e.store.ReconcileRetention(ctx); err != nil { + result.Errors++ + return finish(result, e.cfg.Now()), err + } + } + cursor := "" + dryPayloads := map[string]bool{} + for result.EventsScanned < uint64(e.cfg.MaxEvents) && ctx.Err() == nil { + limit := e.cfg.EventBatchSize + remain := e.cfg.MaxEvents - int(result.EventsScanned) + if limit > remain { + limit = remain + } + page, err := e.store.ScanRetentionEvents(ctx, ScanRequest{Cursor: cursor, Limit: limit}) + if err != nil { + result.Errors++ + return finish(result, e.cfg.Now()), err + } + if len(page.Events) == 0 { + break + } + eligible := make([]Event, 0, len(page.Events)) + times := map[string]time.Time{} + for _, event := range page.Events { + result.EventsScanned++ + expiry, matched := e.eventExpiry(event) + if matched { + result.EventsMatched++ + } + if matched && !expiry.After(now) { + eligible = append(eligible, event) + result.EventsEligible++ + for _, p := range event.Payloads { + if at, ok := e.payloadExpiry(event, p); ok { + k := payloadKey(p.Ref) + if old := times[k]; at.After(old) { + times[k] = at + } + } + } + } + } + if len(eligible) > 0 { + if opt.DryRun { + for key := range times { + dryPayloads[key] = true + } + result.PayloadCandidates = uint64(len(dryPayloads)) + } + archiveID := "" + if !opt.DryRun { + records, err := e.store.LoadArchiveEvents(ctx, eligible) + if err != nil { + result.Errors++ + result.ArchiveFailures++ + return finish(result, e.cfg.Now()), err + } + if !sameArchiveEvents(records, eligible) { + result.Errors++ + result.ArchiveFailures++ + return finish(result, e.cfg.Now()), errors.New("trail/retention: incomplete archive export") + } + receipt, err := e.cfg.Archiver.Write(ctx, records) + if err != nil { + result.Errors++ + result.ArchiveFailures++ + return finish(result, e.cfg.Now()), err + } + if err = e.store.RecordArchive(ctx, receipt); err != nil { + result.Errors++ + result.ArchiveFailures++ + return finish(result, e.cfg.Now()), err + } + archiveID = receipt.Manifest.ID + result.ArchivesCreated++ + result.EventsArchived += uint64(receipt.Manifest.EventCount) + result.PayloadsArchived += uint64(len(receipt.Manifest.Payloads)) + result.ArchiveBytesWritten += uint64(receipt.Manifest.Events.StoredSize + receipt.ManifestRef.StoredSize) + for _, p := range receipt.Manifest.Payloads { + if p.Archive.StoredSize > 0 { + result.ArchiveBytesWritten += uint64(p.Archive.StoredSize) + } + } + } + dr, err := e.store.DeleteRetentionEvents(ctx, DeleteRequest{Events: eligible, PayloadEligibleAt: times, ArchiveID: archiveID, DryRun: opt.DryRun}) + if !opt.DryRun { + result.EventsDeleted += dr.EventsDeleted + result.PayloadCandidates += dr.PayloadCandidates + } + if err != nil { + result.Errors++ + return finish(result, e.cfg.Now()), err + } + } + cursor = page.NextCursor + if !page.HasMore { + break + } + } + if !opt.DryRun { + previousErrors := result.Errors + if err := e.collectPayloads(ctx, now, &result); err != nil { + if result.Errors == previousErrors { + result.Errors++ + } + return finish(result, e.cfg.Now()), err + } + } + return finish(result, e.cfg.Now()), ctx.Err() +} + +func sameArchiveEvents(records []storage.EventRecord, eligible []Event) bool { + if len(records) != len(eligible) { + return false + } + want := make(map[string]bool, len(eligible)) + for _, event := range eligible { + id := event.ID + if len(id) < 32 { + id = strings.Repeat("0", 32-len(id)) + id + } + want[id] = true + } + for _, event := range records { + if !want[hex.EncodeToString(event.ID[:])] { + return false + } + delete(want, hex.EncodeToString(event.ID[:])) + } + return len(want) == 0 +} + +func finish(r Result, now time.Time) Result { + r.FinishedAt = now.UTC() + r.Duration = r.FinishedAt.Sub(r.StartedAt) + return r +} + +func (e *Engine) eventExpiry(v Event) (time.Time, bool) { + if v.RetentionClass != "" { + ttl, ok := e.cfg.EventClasses[v.RetentionClass] + if !ok { + return time.Time{}, false + } + return v.Timestamp.Add(ttl), true + } + for _, p := range e.cfg.EventPolicies { + if matchEvent(p.Match, v) { + return v.Timestamp.Add(p.TTL), true + } + } + if e.cfg.DefaultEventTTL > 0 { + return v.Timestamp.Add(e.cfg.DefaultEventTTL), true + } + return time.Time{}, false +} +func (e *Engine) payloadExpiry(v Event, p Payload) (time.Time, bool) { + if !p.Ref.RetainUntil.IsZero() { + return p.Ref.RetainUntil, true + } + if p.Ref.RetentionClass != "" { + ttl, ok := e.cfg.PayloadClasses[p.Ref.RetentionClass] + if !ok { + return time.Time{}, false + } + return v.Timestamp.Add(ttl), true + } + for _, x := range e.cfg.PayloadPolicies { + if matchEvent(x.Match.EventMatch, v) && (x.Match.Role == "" || x.Match.Role == p.Role) && (x.Match.ContentType == "" || strings.HasPrefix(p.Ref.ContentType, x.Match.ContentType)) { + return v.Timestamp.Add(x.TTL), true + } + } + if e.cfg.DefaultPayloadTTL > 0 { + return v.Timestamp.Add(e.cfg.DefaultPayloadTTL), true + } + return time.Time{}, false +} +func matchEvent(m EventMatch, v Event) bool { + return (m.Kind == "" || m.Kind == v.Kind) && (m.Service == "" || m.Service == v.Service) && (m.Environment == "" || m.Environment == v.Environment) && (m.Level == nil || *m.Level == v.Level) && (m.FlowStatus == "" || m.FlowStatus == v.FlowStatus) && (m.ExecutionStatus == "" || m.ExecutionStatus == v.ExecutionStatus) && (m.HTTPStatusClass == 0 || v.HTTPStatusCode/100 == m.HTTPStatusClass) && (m.HTTPSuccess == nil || *m.HTTPSuccess == v.HTTPSuccess) +} +func payloadKey(r payload.Ref) string { return r.Store + "\x00" + r.Key } + +func (e *Engine) collectPayloads(ctx context.Context, now time.Time, r *Result) error { + remaining := e.cfg.MaxPayloads + for remaining > 0 && ctx.Err() == nil { + limit := e.cfg.PayloadBatchSize + if limit > remaining { + limit = remaining + } + page, err := e.store.ListPayloadCandidates(ctx, now.Add(-e.cfg.PayloadGracePeriod), limit) + if err != nil { + return err + } + if len(page.Candidates) == 0 { + return nil + } + verified := page.Candidates[:0] + for _, candidate := range page.Candidates { + if candidate.ArchiveID == "" { + r.PayloadsUnknown++ + continue + } + verified = append(verified, candidate) + } + page.Candidates = verified + if len(page.Candidates) == 0 { + return nil + } + refs := make([]payload.Ref, len(page.Candidates)) + for i := range page.Candidates { + refs[i] = page.Candidates[i].Ref + } + live, err := e.store.ReferencedPayloads(ctx, refs) + if err != nil { + return err + } + remaining -= len(page.Candidates) + jobs := make(chan Candidate) + var wg sync.WaitGroup + var mu sync.Mutex + var firstErr error + worker := func() { + defer wg.Done() + for c := range jobs { + referenced, orphaned, unknown, deleted, bytesReclaimed, failures, err := e.processCandidate(ctx, c, live[payloadKey(c.Ref)]) + mu.Lock() + if referenced { + r.PayloadsReferenced++ + } + if orphaned { + r.PayloadsOrphaned++ + } + if unknown { + r.PayloadsUnknown++ + } + if deleted { + r.PayloadsDeleted++ + } + r.PayloadBytesReclaimed += bytesReclaimed + r.Errors += failures + if firstErr == nil && err != nil { + firstErr = err + } + mu.Unlock() + } + } + workers := e.cfg.PayloadConcurrency + if workers > len(page.Candidates) { + workers = len(page.Candidates) + } + wg.Add(workers) + for range workers { + go worker() + } + for _, c := range page.Candidates { + jobs <- c + } + close(jobs) + wg.Wait() + if firstErr != nil { + return firstErr + } + if len(page.Candidates) < limit { + return nil + } + } + return ctx.Err() +} + +func (e *Engine) processCandidate(ctx context.Context, c Candidate, live bool) (bool, bool, bool, bool, uint64, uint64, error) { + if live { + return true, false, false, false, 0, 0, e.store.CompletePayloadCandidate(ctx, c.Ref, "referenced") + } + ps := e.payloads[c.Ref.Store] + if ps == nil { + failure := fmt.Errorf("trail/retention: payload store %q is not configured", c.Ref.Store) + if err := e.store.CompletePayloadCandidate(ctx, c.Ref, failure.Error()); err != nil { + return false, false, true, false, 0, 1, err + } + return false, false, true, false, 0, 1, failure + } + if !validContentAddressedRef(c.Ref) { + failure := errors.New("trail/retention: invalid content-addressed payload reference") + if err := e.store.CompletePayloadCandidate(ctx, c.Ref, failure.Error()); err != nil { + return false, false, true, false, 0, 1, err + } + return false, false, true, false, 0, 1, failure + } + err := ps.Delete(ctx, c.Ref) + if errors.Is(err, payload.ErrNotFound) { + err = nil + } + if err != nil { + if recordErr := e.store.CompletePayloadCandidate(ctx, c.Ref, err.Error()); recordErr != nil { + return false, true, false, false, 0, 1, recordErr + } + return false, true, false, false, 0, 1, err + } + if err = e.store.CompletePayloadCandidate(ctx, c.Ref, ""); err != nil { + return false, true, false, true, 0, 0, err + } + var reclaimed uint64 + if c.Ref.StoredSize > 0 { + reclaimed = uint64(c.Ref.StoredSize) + } + return false, true, false, true, reclaimed, 0, nil +} + +func validContentAddressedRef(ref payload.Ref) bool { + var zero [32]byte + if ref.Store == "" || ref.Key == "" || ref.SHA256 == zero { + return false + } + sum := hex.EncodeToString(ref.SHA256[:]) + return strings.HasSuffix(ref.Key, sum) || strings.HasSuffix(ref.Key, sum+".gz") +} diff --git a/retention/retention_test.go b/retention/retention_test.go new file mode 100644 index 0000000..6c5e35d --- /dev/null +++ b/retention/retention_test.go @@ -0,0 +1,325 @@ +package retention + +import ( + "context" + "encoding/hex" + "errors" + "io" + "strings" + "testing" + "time" + + "github.com/vestavision/trail" + "github.com/vestavision/trail/archive" + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/payload/filesystem" + "github.com/vestavision/trail/storage" +) + +type fakeStore struct { + events []Event + candidates []Candidate + deleted int + live map[string]bool + failDelete bool + failRecord bool + lastLimit int +} + +func (*fakeStore) ReconcileRetention(context.Context) error { return nil } +func (*fakeStore) LoadArchiveEvents(_ context.Context, events []Event) ([]storage.EventRecord, error) { + out := make([]storage.EventRecord, len(events)) + for i, event := range events { + raw, err := hex.DecodeString(strings.Repeat("0", 32-len(event.ID)) + event.ID) + if err != nil { + return nil, err + } + copy(out[i].ID[:], raw) + } + return out, nil +} +func (s *fakeStore) RecordArchive(context.Context, archive.Receipt) error { + if s.failRecord { + return errors.New("archive catalog unavailable") + } + return nil +} + +type fakeArchiver struct{} + +func (fakeArchiver) Write(_ context.Context, events []storage.EventRecord) (archive.Receipt, error) { + return archive.Receipt{Manifest: archive.Manifest{ID: "archive", EventCount: len(events)}}, nil +} +func withArchiver(c Config) Config { c.Archiver = fakeArchiver{}; return c } + +type failingArchiver struct{} + +func (failingArchiver) Write(context.Context, []storage.EventRecord) (archive.Receipt, error) { + return archive.Receipt{}, errors.New("archive unavailable") +} + +func (s *fakeStore) ScanRetentionEvents(_ context.Context, r ScanRequest) (EventPage, error) { + s.lastLimit = r.Limit + if r.Cursor != "" { + return EventPage{}, nil + } + n := len(s.events) + more := false + if n > r.Limit { + n = r.Limit + more = true + } + return EventPage{Events: s.events[:n], HasMore: more, NextCursor: "next"}, nil +} +func (s *fakeStore) DeleteRetentionEvents(_ context.Context, r DeleteRequest) (DeleteResult, error) { + if s.failDelete { + return DeleteResult{}, errors.New("delete failed") + } + if r.DryRun { + return DeleteResult{}, nil + } + s.deleted += len(r.Events) + seen := map[string]bool{} + for _, e := range r.Events { + for _, p := range e.Payloads { + k := payloadKey(p.Ref) + if at, ok := r.PayloadEligibleAt[k]; ok && !seen[k] { + s.candidates = append(s.candidates, Candidate{Ref: p.Ref, ArchiveID: r.ArchiveID, EligibleAt: at, CandidateAt: time.Unix(1, 0)}) + seen[k] = true + } + } + } + return DeleteResult{EventsDeleted: uint64(len(r.Events)), PayloadCandidates: uint64(len(seen))}, nil +} +func (s *fakeStore) ListPayloadCandidates(_ context.Context, _ time.Time, n int) (CandidatePage, error) { + if len(s.candidates) > n { + return CandidatePage{Candidates: s.candidates[:n]}, nil + } + return CandidatePage{Candidates: append([]Candidate(nil), s.candidates...)}, nil +} +func (s *fakeStore) ReferencedPayloads(_ context.Context, refs []payload.Ref) (map[string]bool, error) { + out := map[string]bool{} + for _, r := range refs { + out[payloadKey(r)] = s.live[payloadKey(r)] + } + return out, nil +} +func (s *fakeStore) CompletePayloadCandidate(_ context.Context, r payload.Ref, failure string) error { + if failure == "" || failure == "referenced" { + k := payloadKey(r) + for i := range s.candidates { + if payloadKey(s.candidates[i].Ref) == k { + s.candidates = append(s.candidates[:i], s.candidates[i+1:]...) + break + } + } + } + return nil +} +func TestPolicyPrecedenceAndDryRun(t *testing.T) { + now := time.Unix(1_000_000, 0).UTC() + warn := trail.LevelWarn + store := &fakeStore{events: []Event{{ID: "1", Timestamp: now.Add(-48 * time.Hour), Kind: "provider.http", Level: warn, RetentionClass: "audit"}, {ID: "2", Timestamp: now.Add(-48 * time.Hour), Kind: "provider.http", Level: warn}}} + cfg := Config{Now: func() time.Time { return now }, EventClasses: map[string]time.Duration{"audit": 365 * 24 * time.Hour}, EventPolicies: []EventPolicy{{Name: "http", Match: EventMatch{Kind: "provider.http"}, TTL: 24 * time.Hour}}, PayloadGracePeriod: time.Hour} + engine, err := New(withArchiver(cfg), store, nil) + if err != nil { + t.Fatal(err) + } + r, err := engine.Run(context.Background(), RunOptions{DryRun: true}) + if err != nil { + t.Fatal(err) + } + if r.EventsScanned != 2 || r.EventsEligible != 1 || r.EventsDeleted != 0 || store.deleted != 0 { + t.Fatalf("result=%+v deleted=%d", r, store.deleted) + } +} + +func TestUnknownClassesRetainAndOrderedPolicyWins(t *testing.T) { + now := time.Unix(1_500_000, 0).UTC() + e := &Engine{cfg: Config{EventClasses: map[string]time.Duration{"audit": time.Hour}, EventPolicies: []EventPolicy{{Name: "first", Match: EventMatch{Kind: "x"}, TTL: time.Hour}, {Name: "second", Match: EventMatch{Kind: "x"}, TTL: 2 * time.Hour}}, DefaultEventTTL: 3 * time.Hour}} + if _, ok := e.eventExpiry(Event{Timestamp: now, Kind: "x", RetentionClass: "unknown"}); ok { + t.Fatal("unknown class must retain") + } + expiry, ok := e.eventExpiry(Event{Timestamp: now, Kind: "x"}) + if !ok || !expiry.Equal(now.Add(time.Hour)) { + t.Fatalf("expiry=%v ok=%v", expiry, ok) + } + ref := payload.Ref{RetentionClass: "unknown"} + if _, ok = e.payloadExpiry(Event{Timestamp: now}, Payload{Ref: ref}); ok { + t.Fatal("unknown payload class must retain") + } +} + +func TestExplicitPayloadExpiryAndSharedReference(t *testing.T) { + now := time.Unix(2_000_000, 0).UTC() + ref := payload.Ref{Store: "memory", Key: "abc", StoredSize: 42, RetainUntil: now.Add(-time.Hour)} + store := &fakeStore{events: []Event{{ID: "1", Timestamp: now.Add(-48 * time.Hour), Payloads: []Payload{{Role: "response", Ref: ref}}}}, live: map[string]bool{payloadKey(ref): true}} + ps := &memoryPayload{} + engine, err := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: 24 * time.Hour, PayloadGracePeriod: time.Hour}), store, map[string]payload.Store{"memory": ps}) + if err != nil { + t.Fatal(err) + } + r, err := engine.Run(context.Background(), RunOptions{}) + if err != nil { + t.Fatal(err) + } + if r.EventsDeleted != 1 || r.PayloadsReferenced != 1 || ps.deleted != 0 { + t.Fatalf("result=%+v deleted=%d", r, ps.deleted) + } +} + +func TestDryRunReportsPayloadCandidateWithoutStaging(t *testing.T) { + now := time.Unix(2_500_000, 0).UTC() + ref := payload.Ref{Store: "filesystem", Key: "candidate", RetainUntil: now.Add(-time.Hour)} + store := &fakeStore{events: []Event{{ID: "1", Timestamp: now.Add(-48 * time.Hour), Payloads: []Payload{{Role: "response", Ref: ref}}}}} + engine, err := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Hour}), store, nil) + if err != nil { + t.Fatal(err) + } + r, err := engine.Run(context.Background(), RunOptions{DryRun: true}) + if err != nil { + t.Fatal(err) + } + if r.PayloadCandidates != 1 || len(store.candidates) != 0 || r.EventsDeleted != 0 { + t.Fatalf("result=%+v candidates=%d", r, len(store.candidates)) + } +} + +func TestDeleteFailureDoesNotTouchPayload(t *testing.T) { + now := time.Unix(3_000_000, 0).UTC() + store := &fakeStore{failDelete: true, events: []Event{{ID: "1", Timestamp: now.Add(-48 * time.Hour)}}} + engine, _ := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Hour}), store, nil) + r, err := engine.Run(context.Background(), RunOptions{}) + if err == nil || r.EventsDeleted != 0 || len(store.candidates) != 0 { + t.Fatalf("result=%+v err=%v", r, err) + } +} + +func TestArchiveFailureAndCatalogFailurePreventDeletion(t *testing.T) { + now := time.Unix(3_100_000, 0).UTC() + for _, tc := range []struct { + name string + archiver archive.Writer + failRecord bool + }{{"object write", failingArchiver{}, false}, {"catalog commit", fakeArchiver{}, true}} { + t.Run(tc.name, func(t *testing.T) { + store := &fakeStore{failRecord: tc.failRecord, events: []Event{{ID: "1", Timestamp: now.Add(-48 * time.Hour)}}} + cfg := Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Hour, Archiver: tc.archiver} + engine, err := New(cfg, store, nil) + if err != nil { + t.Fatal(err) + } + result, err := engine.Run(context.Background(), RunOptions{}) + if err == nil || store.deleted != 0 || result.EventsDeleted != 0 || result.ArchiveFailures != 1 { + t.Fatalf("result=%+v deleted=%d err=%v", result, store.deleted, err) + } + }) + } +} + +func TestArchiveIsMandatory(t *testing.T) { + _, err := New(Config{DefaultEventTTL: time.Hour}, &fakeStore{}, nil) + if err == nil || !strings.Contains(err.Error(), "archiver is required") { + t.Fatalf("err=%v", err) + } +} + +func TestRunHonorsEventLimitAndCancellation(t *testing.T) { + now := time.Unix(3_500_000, 0).UTC() + store := &fakeStore{events: []Event{ + {ID: "1", Timestamp: now.Add(-time.Hour)}, + {ID: "2", Timestamp: now.Add(-time.Hour)}, + {ID: "3", Timestamp: now.Add(-time.Hour)}, + }} + engine, err := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Minute, EventBatchSize: 10, MaxEvents: 2}), store, nil) + if err != nil { + t.Fatal(err) + } + r, err := engine.Run(context.Background(), RunOptions{DryRun: true}) + if err != nil { + t.Fatal(err) + } + if r.EventsScanned != 2 || store.lastLimit != 2 { + t.Fatalf("result=%+v limit=%d", r, store.lastLimit) + } + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + store.deleted = 0 + _, err = engine.Run(cancelled, RunOptions{}) + if !errors.Is(err, context.Canceled) || store.deleted != 0 { + t.Fatalf("cancel err=%v deleted=%d", err, store.deleted) + } +} + +func TestFilesystemPayloadCleanup(t *testing.T) { + now := time.Unix(4_000_000, 0).UTC() + fs, err := filesystem.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + ref, err := fs.Put(context.Background(), strings.NewReader("expired payload"), payload.PutOptions{}) + if err != nil { + t.Fatal(err) + } + store := &fakeStore{candidates: []Candidate{{Ref: ref, ArchiveID: "archive", EligibleAt: now.Add(-2 * time.Hour), CandidateAt: now.Add(-2 * time.Hour)}}, live: map[string]bool{}} + engine, err := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Hour, PayloadGracePeriod: time.Hour}), store, map[string]payload.Store{"filesystem": fs}) + if err != nil { + t.Fatal(err) + } + r, err := engine.Run(context.Background(), RunOptions{}) + if err != nil { + t.Fatal(err) + } + if r.PayloadsDeleted != 1 || r.PayloadBytesReclaimed == 0 { + t.Fatalf("result=%+v", r) + } + if _, err = fs.Open(context.Background(), ref); !errors.Is(err, payload.ErrNotFound) { + t.Fatalf("open deleted payload: %v", err) + } +} + +func TestPayloadDeleteFailureRemainsRetryable(t *testing.T) { + now := time.Unix(4_500_000, 0).UTC() + var sum [32]byte + sum[0] = 1 + key := hex.EncodeToString(sum[:]) + ref := payload.Ref{Store: "memory", Key: key, SHA256: sum, StoredSize: 9} + store := &fakeStore{candidates: []Candidate{{Ref: ref, ArchiveID: "archive", EligibleAt: now.Add(-2 * time.Hour), CandidateAt: now.Add(-2 * time.Hour)}}, live: map[string]bool{}} + ps := &memoryPayload{err: errors.New("object store unavailable")} + engine, err := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Hour, PayloadGracePeriod: time.Hour}), store, map[string]payload.Store{"memory": ps}) + if err != nil { + t.Fatal(err) + } + r, err := engine.Run(context.Background(), RunOptions{}) + if err == nil || r.Errors != 1 || r.PayloadsDeleted != 0 || len(store.candidates) != 1 { + t.Fatalf("result=%+v err=%v candidates=%d", r, err, len(store.candidates)) + } +} + +func TestPayloadCandidateWithoutArchiveIsNeverDeleted(t *testing.T) { + now := time.Unix(4_700_000, 0).UTC() + ref := payload.Ref{Store: "memory", Key: "legacy", StoredSize: 9} + store := &fakeStore{candidates: []Candidate{{Ref: ref, EligibleAt: now.Add(-2 * time.Hour), CandidateAt: now.Add(-2 * time.Hour)}}} + ps := &memoryPayload{} + engine, err := New(withArchiver(Config{Now: func() time.Time { return now }, DefaultEventTTL: time.Hour, PayloadGracePeriod: time.Hour}), store, map[string]payload.Store{"memory": ps}) + if err != nil { + t.Fatal(err) + } + result, err := engine.Run(context.Background(), RunOptions{}) + if err != nil || ps.deleted != 0 || result.PayloadsUnknown != 1 { + t.Fatalf("result=%+v deleted=%d err=%v", result, ps.deleted, err) + } +} + +type memoryPayload struct { + deleted int + err error +} + +func (*memoryPayload) Put(context.Context, io.Reader, payload.PutOptions) (payload.Ref, error) { + panic("unused") +} +func (*memoryPayload) Open(context.Context, payload.Ref) (io.ReadCloser, error) { + return nil, payload.ErrNotFound +} +func (p *memoryPayload) Delete(context.Context, payload.Ref) error { p.deleted++; return p.err } diff --git a/storage/clickhouse/migrations.go b/storage/clickhouse/migrations.go index d7fa1a4..0f272a7 100644 --- a/storage/clickhouse/migrations.go +++ b/storage/clickhouse/migrations.go @@ -168,6 +168,20 @@ WHERE flow_id != unhex('00000000000000000000000000000000') var migrationStatements = []string{ createEventsTable, + `CREATE TABLE IF NOT EXISTS trail_payload_gc_candidates ( + store LowCardinality(String), object_key String, ref_json String, + eligible_at DateTime64(9, 'UTC'), candidate_at DateTime64(9, 'UTC'), + retry_after DateTime64(9, 'UTC'), attempts UInt32, last_error String, archive_id String DEFAULT '', + version DateTime64(9, 'UTC') + ) ENGINE=ReplacingMergeTree(version) ORDER BY (store,object_key)`, + `ALTER TABLE trail_payload_gc_candidates ADD COLUMN IF NOT EXISTS archive_id String DEFAULT ''`, + `CREATE TABLE IF NOT EXISTS trail_retention_summary_rebuilds ( + kind Enum8('flow'=1,'execution'=2), identity FixedString(16), version DateTime64(9,'UTC') + ) ENGINE=ReplacingMergeTree(version) ORDER BY (kind,identity)`, + `CREATE TABLE IF NOT EXISTS trail_archives (archive_id String, manifest_json String, manifest_ref_json String, created_at DateTime64(9,'UTC'), version DateTime64(9,'UTC')) ENGINE=ReplacingMergeTree(version) ORDER BY archive_id`, + `CREATE TABLE IF NOT EXISTS trail_event_archives (event_id FixedString(16), archive_id String, archived_at DateTime64(9,'UTC'), version DateTime64(9,'UTC')) ENGINE=ReplacingMergeTree(version) ORDER BY event_id SETTINGS deduplicate_merge_projection_mode='rebuild'`, + `ALTER TABLE trail_event_archives MODIFY SETTING deduplicate_merge_projection_mode='rebuild'`, + `ALTER TABLE trail_event_archives ADD PROJECTION IF NOT EXISTS by_archive (SELECT _part_offset ORDER BY (archive_id,event_id))`, createFlowSummaries, createFlowSummaryView, createExecutionSummaries, diff --git a/storage/clickhouse/retention.go b/storage/clickhouse/retention.go new file mode 100644 index 0000000..e96afe3 --- /dev/null +++ b/storage/clickhouse/retention.go @@ -0,0 +1,530 @@ +package clickhouse + +import ( + "context" + "encoding/hex" + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + "github.com/vestavision/trail" + "github.com/vestavision/trail/archive" + "github.com/vestavision/trail/convention" + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/retention" + "github.com/vestavision/trail/storage" + "github.com/vestavision/trail/wire" +) + +func (s *Store) ScanRetentionEvents(ctx context.Context, req retention.ScanRequest) (retention.EventPage, error) { + if req.Limit <= 0 { + return retention.EventPage{}, nil + } + q := `SELECT event_time,event_id,flow_id,execution_id,kind,toString(level),service,environment,flow_status,execution_status,http_status_code,http_success,field_keys,field_types,field_text,field_num,payload_refs_json FROM trail_events FINAL` + args := []any{} + if req.Cursor != "" { + ts, id, err := parseCHRetentionCursor(req.Cursor) + if err != nil { + return retention.EventPage{}, err + } + q += ` WHERE (event_time,event_id)>(?,?)` + args = []any{ts, string(id)} + } + q += ` ORDER BY event_time,event_id LIMIT ?` + args = append(args, req.Limit+1) + rows, err := s.conn.Query(ctx, q, args...) + if err != nil { + return retention.EventPage{}, err + } + defer rows.Close() + var out retention.EventPage + for rows.Next() { + var v retention.Event + var id, flow, execution, level, payloadJSON string + var keys, types, texts []string + var nums []uint64 + var status uint16 + var success uint8 + if err = rows.Scan(&v.Timestamp, &id, &flow, &execution, &v.Kind, &level, &v.Service, &v.Environment, &v.FlowStatus, &v.ExecutionStatus, &status, &success, &keys, &types, &texts, &nums, &payloadJSON); err != nil { + return out, err + } + v.ID = hex.EncodeToString([]byte(id)) + v.FlowID = hex.EncodeToString([]byte(flow)) + v.ExecutionID = hex.EncodeToString([]byte(execution)) + v.Cursor = chRetentionCursor(v.Timestamp, []byte(id)) + v.Level = chRetentionLevel(level) + v.HTTPStatusCode = int(status) + v.HTTPSuccess = success != 0 + for i, k := range keys { + if k == convention.FieldRetentionClass && i < len(texts) { + v.RetentionClass = texts[i] + break + } + } + var links []storage.PayloadLink + if err = json.Unmarshal([]byte(payloadJSON), &links); err != nil { + return out, err + } + for _, p := range links { + v.Payloads = append(v.Payloads, retention.Payload{Role: p.Role, Ref: p.Ref}) + } + out.Events = append(out.Events, v) + } + if err = rows.Err(); err != nil { + return out, err + } + if len(out.Events) > req.Limit { + out.Events = out.Events[:req.Limit] + out.HasMore = true + } + if len(out.Events) > 0 { + out.NextCursor = out.Events[len(out.Events)-1].Cursor + } + return out, nil +} + +func (s *Store) DeleteRetentionEvents(ctx context.Context, req retention.DeleteRequest) (retention.DeleteResult, error) { + var out retention.DeleteResult + if len(req.Events) == 0 { + return out, nil + } + if req.DryRun { + return out, nil + } + if req.ArchiveID == "" { + return out, fmt.Errorf("trail/clickhouse: archive receipt is required") + } + ids := make([]string, 0, len(req.Events)) + for _, e := range req.Events { + id, er := hex.DecodeString(e.ID) + if er != nil { + return out, er + } + ids = append(ids, string(id)) + } + var archived uint64 + if err := s.conn.QueryRow(ctx, `SELECT count() FROM trail_event_archives FINAL WHERE archive_id=? AND event_id IN ?`, req.ArchiveID, ids).Scan(&archived); err != nil { + return out, err + } + if archived != uint64(len(ids)) { + return out, fmt.Errorf("trail/clickhouse: archive receipt does not cover deletion batch") + } + now := time.Now().UTC() + seen := map[string]bool{} + type staged struct { + ref payload.Ref + eligible, candidate time.Time + } + var stagedItems []staged + for _, e := range req.Events { + for _, p := range e.Payloads { + k := p.Ref.Store + "\x00" + p.Ref.Key + at, ok := req.PayloadEligibleAt[k] + if !ok || seen[k] || p.Ref.Store == "" || p.Ref.Key == "" { + continue + } + seen[k] = true + candidateAt := now + var count uint64 + var oldEligible, oldCandidate time.Time + if er := s.conn.QueryRow(ctx, `SELECT count(),max(eligible_at),min(candidate_at) FROM trail_payload_gc_candidates FINAL WHERE store=? AND object_key=?`, p.Ref.Store, p.Ref.Key).Scan(&count, &oldEligible, &oldCandidate); er != nil { + return out, er + } + if count > 0 { + if oldEligible.After(at) { + at = oldEligible + } + if oldCandidate.Before(candidateAt) { + candidateAt = oldCandidate + } + } + stagedItems = append(stagedItems, staged{ref: p.Ref, eligible: at, candidate: candidateAt}) + } + } + batch, err := s.conn.PrepareBatch(ctx, `INSERT INTO trail_payload_gc_candidates (store,object_key,ref_json,eligible_at,candidate_at,retry_after,attempts,last_error,archive_id,version)`) + if err != nil { + return out, err + } + for _, item := range stagedItems { + raw, er := json.Marshal(item.ref) + if er != nil { + _ = batch.Abort() + return out, er + } + if er = batch.Append(item.ref.Store, item.ref.Key, string(raw), item.eligible, item.candidate, now, uint32(0), "", req.ArchiveID, now); er != nil { + _ = batch.Abort() + return out, er + } + out.PayloadCandidates++ + } + if out.PayloadCandidates > 0 { + if err = batch.Send(); err != nil { + return retention.DeleteResult{}, err + } + } else { + _ = batch.Abort() + } + if err = s.stageRetentionSummaryRebuilds(ctx, req.Events); err != nil { + return out, err + } + if err = s.conn.Exec(ctx, `ALTER TABLE trail_events DELETE WHERE event_id IN ? SETTINGS mutations_sync=2`, ids); err != nil { + return out, err + } + out.EventsDeleted = uint64(len(ids)) + if err = s.ReconcileRetention(ctx); err != nil { + return out, err + } + return out, nil +} + +func (s *Store) LoadArchiveEvents(ctx context.Context, events []retention.Event) ([]storage.EventRecord, error) { + ids := make([]string, 0, len(events)) + for _, e := range events { + id, err := hex.DecodeString(e.ID) + if err != nil { + return nil, err + } + ids = append(ids, string(id)) + } + rows, err := s.conn.Query(ctx, eventRows+` FINAL WHERE event_id IN ? ORDER BY event_time,event_id`, ids) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]storage.EventRecord, 0, len(events)) + for rows.Next() { + v, err := scanCHEvent(rows) + if err != nil { + return nil, err + } + out = append(out, v) + } + return out, rows.Err() +} +func (s *Store) RecordArchive(ctx context.Context, receipt archive.Receipt) error { + manifest, err := json.Marshal(receipt.Manifest) + if err != nil { + return err + } + ref, err := json.Marshal(receipt.ManifestRef) + if err != nil { + return err + } + now := time.Now().UTC() + if err = s.conn.Exec(ctx, `INSERT INTO trail_archives VALUES(?,?,?,?,?)`, receipt.Manifest.ID, string(manifest), string(ref), receipt.Manifest.CreatedAt, now); err != nil { + return err + } + batch, err := s.conn.PrepareBatch(ctx, `INSERT INTO trail_event_archives`) + if err != nil { + return err + } + for _, id := range receipt.Manifest.EventIDs { + parsed, er := trail.ParseEventID(id) + if er != nil { + _ = batch.Abort() + return er + } + if er = batch.Append(string(parsed[:]), receipt.Manifest.ID, now, now); er != nil { + _ = batch.Abort() + return er + } + } + return batch.Send() +} + +func (s *Store) ListArchives(ctx context.Context, offset, limit int) (archive.Page, error) { + if offset < 0 || limit < 1 || limit > 500 { + return archive.Page{}, storage.ErrInvalidCursor + } + rows, err := s.conn.Query(ctx, `SELECT manifest_json,manifest_ref_json FROM trail_archives FINAL ORDER BY created_at DESC,archive_id LIMIT ? OFFSET ?`, limit+1, offset) + if err != nil { + return archive.Page{}, err + } + defer rows.Close() + var out archive.Page + for rows.Next() { + var manifestJSON, refJSON string + if err = rows.Scan(&manifestJSON, &refJSON); err != nil { + return out, err + } + var receipt archive.Receipt + if err = json.Unmarshal([]byte(manifestJSON), &receipt.Manifest); err != nil { + return out, err + } + if err = json.Unmarshal([]byte(refJSON), &receipt.ManifestRef); err != nil { + return out, err + } + out.Items = append(out.Items, receipt) + } + if len(out.Items) > limit { + out.Items = out.Items[:limit] + out.HasMore = true + } + return out, rows.Err() +} + +func (s *Store) GetArchive(ctx context.Context, id string) (archive.Receipt, error) { + var manifestJSON, refJSON string + var count uint64 + if err := s.conn.QueryRow(ctx, `SELECT count(),any(manifest_json),any(manifest_ref_json) FROM trail_archives FINAL WHERE archive_id=?`, id).Scan(&count, &manifestJSON, &refJSON); err != nil { + return archive.Receipt{}, err + } + if count == 0 { + return archive.Receipt{}, storage.ErrNotFound + } + var out archive.Receipt + if err := json.Unmarshal([]byte(manifestJSON), &out.Manifest); err != nil { + return out, err + } + if err := json.Unmarshal([]byte(refJSON), &out.ManifestRef); err != nil { + return out, err + } + return out, nil +} + +// RestoreArchiveEvents uses an archive-specific insertion identity. The +// ReplacingMergeTree event_id key makes repeated restores idempotent after +// merges, while the token makes immediate retries safe. +func (s *Store) RestoreArchiveEvents(ctx context.Context, archiveID string, events []storage.EventRecord) (int, error) { + if len(events) == 0 { + return 0, nil + } + raw, err := hex.DecodeString(archiveID) + if err != nil || len(raw) < 16 { + return 0, fmt.Errorf("trail/clickhouse: invalid archive ID") + } + var batchID trail.EventID + copy(batchID[:], raw[:16]) + for i := range events { + events[i].BatchID = batchID + events[i].Delivery.Stream = "trail-archive-restore" + } + if err = s.WriteBatches(ctx, []storage.Batch{{BatchID: batchID, Delivery: storage.Delivery{Stream: "trail-archive-restore"}, Events: events}}); err != nil { + return 0, err + } + return len(events), nil +} + +func (s *Store) stageRetentionSummaryRebuilds(ctx context.Context, events []retention.Event) error { + now := time.Now().UTC() + batch, err := s.conn.PrepareBatch(ctx, `INSERT INTO trail_retention_summary_rebuilds`) + if err != nil { + return err + } + seen := map[string]bool{} + count := 0 + for _, e := range events { + for _, v := range []struct{ kind, id string }{{"flow", e.FlowID}, {"execution", e.ExecutionID}} { + if v.id == "" || v.id == strings.Repeat("0", 32) || seen[v.kind+v.id] { + continue + } + id, er := hex.DecodeString(v.id) + if er != nil { + _ = batch.Abort() + return er + } + if er = batch.Append(v.kind, string(id), now); er != nil { + _ = batch.Abort() + return er + } + seen[v.kind+v.id] = true + count++ + } + } + if count == 0 { + _ = batch.Abort() + return nil + } + return batch.Send() +} + +func (s *Store) ReconcileRetention(ctx context.Context) error { + rows, err := s.conn.Query(ctx, `SELECT kind,identity FROM trail_retention_summary_rebuilds FINAL ORDER BY kind,identity LIMIT 1000`) + if err != nil { + return err + } + var events []retention.Event + var taskIDs []string + for rows.Next() { + var kind, id string + if err = rows.Scan(&kind, &id); err != nil { + rows.Close() + return err + } + hexID := hex.EncodeToString([]byte(id)) + e := retention.Event{} + if kind == "flow" { + e.FlowID = hexID + } else { + e.ExecutionID = hexID + } + events = append(events, e) + taskIDs = append(taskIDs, id) + } + if err = rows.Err(); err != nil { + rows.Close() + return err + } + rows.Close() + if len(events) == 0 { + return nil + } + if err = s.rebuildRetentionSummaries(ctx, events); err != nil { + return err + } + return s.conn.Exec(ctx, `ALTER TABLE trail_retention_summary_rebuilds DELETE WHERE identity IN ? SETTINGS mutations_sync=2`, taskIDs) +} + +func (s *Store) rebuildRetentionSummaries(ctx context.Context, events []retention.Event) error { + zero := strings.Repeat("0", 32) + flowSet, executionSet := map[string]bool{}, map[string]bool{} + for _, e := range events { + if e.FlowID != "" && e.FlowID != zero { + b, err := hex.DecodeString(e.FlowID) + if err != nil { + return err + } + flowSet[string(b)] = true + } + if e.ExecutionID != "" && e.ExecutionID != zero { + b, err := hex.DecodeString(e.ExecutionID) + if err != nil { + return err + } + executionSet[string(b)] = true + } + } + flows := make([]string, 0, len(flowSet)) + for id := range flowSet { + flows = append(flows, id) + } + executions := make([]string, 0, len(executionSet)) + for id := range executionSet { + executions = append(executions, id) + } + if len(flows) > 0 { + if err := s.conn.Exec(ctx, `ALTER TABLE trail_flow_summaries DELETE WHERE flow_id IN ? SETTINGS mutations_sync=2`, flows); err != nil { + return err + } + if err := s.conn.Exec(ctx, `INSERT INTO trail_flow_summaries SELECT flow_id,service,environment,minState(event_time),maxState(event_time),uniqExactState(event_id),argMaxState(entity_type,tuple(event_time,event_id)),argMaxState(entity_id,tuple(event_time,event_id)),argMaxState(execution_id,tuple(event_time,event_id)),argMaxIfState(flow_status,tuple(event_time,event_id),notEmpty(flow_status)),argMinState(kind,tuple(event_time,event_id)),argMaxState(kind,tuple(event_time,event_id)) FROM trail_events WHERE flow_id IN ? GROUP BY flow_id,service,environment`, flows); err != nil { + return err + } + if err := s.conn.Exec(ctx, `ALTER TABLE trail_flow_terminals DELETE WHERE flow_id IN ? SETTINGS mutations_sync=2`, flows); err != nil { + return err + } + if err := s.conn.Exec(ctx, `INSERT INTO trail_flow_terminals SELECT flow_id,event_time,service,environment,entity_type,entity_id,execution_id,flow_status FROM trail_events WHERE flow_id IN ? AND flow_status IN ('succeeded','failed','cancelled')`, flows); err != nil { + return err + } + } + if len(executions) > 0 { + if err := s.conn.Exec(ctx, `ALTER TABLE trail_execution_summaries DELETE WHERE execution_id IN ? SETTINGS mutations_sync=2`, executions); err != nil { + return err + } + if err := s.conn.Exec(ctx, `INSERT INTO trail_execution_summaries SELECT execution_id,service,environment,minState(event_time),maxState(event_time),uniqExactState(event_id),uniqExactStateIf(flow_id,flow_id!=unhex('00000000000000000000000000000000')),argMaxIfState(execution_kind,tuple(event_time,event_id),notEmpty(execution_kind)),argMaxIfState(execution_source,tuple(event_time,event_id),notEmpty(execution_source)),argMaxIfState(execution_status,tuple(event_time,event_id),notEmpty(execution_status)),maxState(execution_attempt),argMaxState(parent_execution_id,tuple(event_time,event_id)),argMaxState(retry_of_execution_id,tuple(event_time,event_id)) FROM trail_events WHERE execution_id IN ? GROUP BY execution_id,service,environment`, executions); err != nil { + return err + } + } + return nil +} + +func (s *Store) ListPayloadCandidates(ctx context.Context, before time.Time, limit int) (retention.CandidatePage, error) { + rows, err := s.conn.Query(ctx, `SELECT ref_json,archive_id,eligible_at,candidate_at,attempts,last_error FROM trail_payload_gc_candidates FINAL WHERE archive_id!='' AND eligible_at<=now64(9) AND candidate_at<=? AND retry_after<=now64(9) ORDER BY candidate_at,store,object_key LIMIT ?`, before, limit) + if err != nil { + return retention.CandidatePage{}, err + } + defer rows.Close() + var out retention.CandidatePage + for rows.Next() { + var raw string + var c retention.Candidate + if err = rows.Scan(&raw, &c.ArchiveID, &c.EligibleAt, &c.CandidateAt, &c.Attempts, &c.LastError); err != nil { + return out, err + } + if err = json.Unmarshal([]byte(raw), &c.Ref); err != nil { + return out, err + } + out.Candidates = append(out.Candidates, c) + } + return out, rows.Err() +} +func (s *Store) ReferencedPayloads(ctx context.Context, refs []payload.Ref) (map[string]bool, error) { + out := map[string]bool{} + for _, ref := range refs { + rows, err := s.conn.Query(ctx, `SELECT payload_refs_json FROM trail_events WHERE position(payload_refs_json,?)>0`, ref.Key) + if err != nil { + return nil, err + } + found := false + for rows.Next() { + var raw string + if err = rows.Scan(&raw); err != nil { + rows.Close() + return nil, err + } + var links []storage.PayloadLink + if json.Unmarshal([]byte(raw), &links) == nil { + for _, p := range links { + if p.Ref.Store == ref.Store && p.Ref.Key == ref.Key { + found = true + break + } + } + } + if found { + break + } + } + rows.Close() + out[ref.Store+"\x00"+ref.Key] = found + } + return out, nil +} +func (s *Store) CompletePayloadCandidate(ctx context.Context, ref payload.Ref, failure string) error { + if failure == "" || failure == "referenced" { + return s.conn.Exec(ctx, `ALTER TABLE trail_payload_gc_candidates DELETE WHERE store=? AND object_key=? SETTINGS mutations_sync=2`, ref.Store, ref.Key) + } + var raw string + var archiveID string + var eligible, candidate time.Time + var attempts uint32 + err := s.conn.QueryRow(ctx, `SELECT ref_json,archive_id,eligible_at,candidate_at,attempts FROM trail_payload_gc_candidates FINAL WHERE store=? AND object_key=?`, ref.Store, ref.Key).Scan(&raw, &archiveID, &eligible, &candidate, &attempts) + if err != nil { + return err + } + now := time.Now().UTC() + return s.conn.Exec(ctx, `INSERT INTO trail_payload_gc_candidates (store,object_key,ref_json,eligible_at,candidate_at,retry_after,attempts,last_error,archive_id,version) VALUES(?,?,?,?,?,?,?,?,?,?)`, ref.Store, ref.Key, raw, eligible, candidate, now.Add(time.Hour), attempts+1, failure, archiveID, now) +} +func chRetentionCursor(t time.Time, id []byte) string { + return strconv.FormatInt(t.UnixNano(), 10) + ":" + hex.EncodeToString(id) +} +func parseCHRetentionCursor(v string) (time.Time, []byte, error) { + a, b, ok := strings.Cut(v, ":") + if !ok { + return time.Time{}, nil, fmt.Errorf("invalid retention cursor") + } + n, err := strconv.ParseInt(a, 10, 64) + if err != nil { + return time.Time{}, nil, err + } + id, err := hex.DecodeString(b) + return time.Unix(0, n).UTC(), id, err +} +func chRetentionLevel(v string) trail.Level { + switch v { + case "debug": + return trail.LevelDebug + case "warn": + return trail.LevelWarn + case "error": + return trail.LevelError + default: + return trail.LevelInfo + } +} + +var _ retention.Store = (*Store)(nil) +var _ archive.Catalog = (*Store)(nil) +var _ archive.RestoreStore = (*Store)(nil) +var _ = wire.Field{} diff --git a/storage/clickhouse/retention_test.go b/storage/clickhouse/retention_test.go new file mode 100644 index 0000000..278a9ac --- /dev/null +++ b/storage/clickhouse/retention_test.go @@ -0,0 +1,36 @@ +package clickhouse + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/vestavision/trail/retention" +) + +func TestRetentionCursorRoundTrip(t *testing.T) { + at := time.Unix(123, 456).UTC() + id := bytes.Repeat([]byte{9}, 16) + cursor := chRetentionCursor(at, id) + gotAt, gotID, err := parseCHRetentionCursor(cursor) + if err != nil { + t.Fatal(err) + } + if !gotAt.Equal(at) || !bytes.Equal(gotID, id) { + t.Fatalf("at=%v id=%x", gotAt, gotID) + } +} + +func TestDeleteRequiresArchiveReceiptBeforeDatabaseAccess(t *testing.T) { + store := &Store{} + _, err := store.DeleteRetentionEvents(context.Background(), retention.DeleteRequest{Events: []retention.Event{{ID: "00"}}}) + if err == nil { + t.Fatal("expected archive receipt error") + } +} +func TestRetentionCursorRejectsInvalid(t *testing.T) { + if _, _, err := parseCHRetentionCursor("invalid"); err == nil { + t.Fatal("expected invalid cursor error") + } +} diff --git a/storage/normalize.go b/storage/normalize.go index 31d44f5..50f57ec 100644 --- a/storage/normalize.go +++ b/storage/normalize.go @@ -64,6 +64,8 @@ func normalizeEvent(event wire.Event, metadata trail.Metadata, batchID trail.Eve record.ExecutionStatus = field.Text case convention.FieldExecutionKind: record.ExecutionKind = field.Text + case convention.FieldRetentionClass: + record.RetentionClass = field.Text case "http.method": record.HTTP.Method = field.Text case "http.scheme": diff --git a/storage/normalize_test.go b/storage/normalize_test.go index b93f5a3..f5599aa 100644 --- a/storage/normalize_test.go +++ b/storage/normalize_test.go @@ -26,6 +26,7 @@ func TestNormalizePromotesPortableFields(t *testing.T) { Fields: []wire.Field{ {Key: convention.FieldFlowStatus, Type: "string", Text: "awaiting_review"}, {Key: convention.FieldExecutionStatus, Type: "string", Text: string(convention.StatusFailed)}, + {Key: convention.FieldRetentionClass, Type: "string", Text: "audit"}, {Key: "http.status_code", Type: "int", Num: 503}, {Key: "http.success", Type: "bool"}, {Key: "payload.response.store", Type: "string", Text: "filesystem"}, @@ -41,7 +42,7 @@ func TestNormalizePromotesPortableFields(t *testing.T) { t.Fatal(err) } record := batch.Events[0] - if record.FlowStatus != "awaiting_review" || record.ExecutionStatus != "failed" || record.HTTP.StatusCode != 503 || record.HTTP.Success { + if record.FlowStatus != "awaiting_review" || record.ExecutionStatus != "failed" || record.RetentionClass != "audit" || record.HTTP.StatusCode != 503 || record.HTTP.Success { t.Fatalf("record=%+v", record) } if len(record.Payloads) != 1 || record.Payloads[0].Role != "response" || !record.Payloads[0].Truncated || record.Payloads[0].Ref.SHA256 != [32]byte(checksum) { diff --git a/storage/postgres/migrations.go b/storage/postgres/migrations.go index 4832c66..473608f 100644 --- a/storage/postgres/migrations.go +++ b/storage/postgres/migrations.go @@ -46,6 +46,17 @@ CREATE TABLE IF NOT EXISTS trail_events ( var migrationStatements = []string{ createEventsTable, + `CREATE TABLE IF NOT EXISTS trail_payload_gc_candidates ( + store text NOT NULL, object_key text NOT NULL, ref jsonb NOT NULL, + eligible_at timestamptz NOT NULL, candidate_at timestamptz NOT NULL DEFAULT now(), + retry_after timestamptz NOT NULL DEFAULT now(), attempts integer NOT NULL DEFAULT 0, + last_error text NOT NULL DEFAULT '', archive_id text NOT NULL DEFAULT '', PRIMARY KEY (store, object_key) + )`, + `ALTER TABLE trail_payload_gc_candidates ADD COLUMN IF NOT EXISTS archive_id text NOT NULL DEFAULT ''`, + `CREATE INDEX IF NOT EXISTS trail_payload_gc_ready_idx ON trail_payload_gc_candidates (eligible_at, candidate_at, retry_after)`, + `CREATE TABLE IF NOT EXISTS trail_archives (archive_id text PRIMARY KEY, manifest jsonb NOT NULL, manifest_ref jsonb NOT NULL, created_at timestamptz NOT NULL)`, + `CREATE TABLE IF NOT EXISTS trail_event_archives (event_id bytea PRIMARY KEY, archive_id text NOT NULL REFERENCES trail_archives(archive_id), archived_at timestamptz NOT NULL DEFAULT now())`, + `CREATE INDEX IF NOT EXISTS trail_event_archives_archive_idx ON trail_event_archives(archive_id)`, `CREATE INDEX IF NOT EXISTS trail_events_time_idx ON trail_events (event_time DESC, event_id DESC)`, `CREATE INDEX IF NOT EXISTS trail_events_flow_idx ON trail_events (flow_id, event_time, event_id) WHERE flow_id IS NOT NULL`, `CREATE INDEX IF NOT EXISTS trail_events_execution_idx ON trail_events (execution_id, event_time, event_id) WHERE execution_id IS NOT NULL`, diff --git a/storage/postgres/retention.go b/storage/postgres/retention.go new file mode 100644 index 0000000..9ecd94e --- /dev/null +++ b/storage/postgres/retention.go @@ -0,0 +1,336 @@ +package postgres + +import ( + "context" + "encoding/hex" + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + "github.com/vestavision/trail" + "github.com/vestavision/trail/archive" + "github.com/vestavision/trail/convention" + "github.com/vestavision/trail/payload" + "github.com/vestavision/trail/retention" + "github.com/vestavision/trail/storage" + "github.com/vestavision/trail/wire" +) + +func (s *Store) ScanRetentionEvents(ctx context.Context, req retention.ScanRequest) (retention.EventPage, error) { + if req.Limit <= 0 { + return retention.EventPage{}, nil + } + args := []any{} + where := "" + if req.Cursor != "" { + ts, id, err := parseRetentionCursor(req.Cursor) + if err != nil { + return retention.EventPage{}, err + } + args = []any{ts, id} + where = " WHERE (event_time,event_id)>($1,$2)" + } + args = append(args, req.Limit+1) + q := `SELECT event_time,event_id,flow_id,execution_id,kind,level,service,environment,flow_status,execution_status,http_status_code,http_success,fields,payload_refs FROM trail_events` + where + fmt.Sprintf(` ORDER BY event_time,event_id LIMIT $%d`, len(args)) + rows, err := s.pool.Query(ctx, q, args...) + if err != nil { + return retention.EventPage{}, err + } + defer rows.Close() + var out retention.EventPage + for rows.Next() { + var v retention.Event + var id, flow, execution, fields, payloads []byte + var level string + if err = rows.Scan(&v.Timestamp, &id, &flow, &execution, &v.Kind, &level, &v.Service, &v.Environment, &v.FlowStatus, &v.ExecutionStatus, &v.HTTPStatusCode, &v.HTTPSuccess, &fields, &payloads); err != nil { + return out, err + } + v.ID = hex.EncodeToString(id) + v.FlowID = hex.EncodeToString(flow) + v.ExecutionID = hex.EncodeToString(execution) + v.Cursor = retentionCursor(v.Timestamp, id) + v.Level = retentionLevel(level) + var links []storage.PayloadLink + if err = json.Unmarshal(payloads, &links); err != nil { + return out, err + } + for _, p := range links { + v.Payloads = append(v.Payloads, retention.Payload{Role: p.Role, Ref: p.Ref}) + } + var fs []wire.Field + if json.Unmarshal(fields, &fs) == nil { + for _, f := range fs { + if f.Key == convention.FieldRetentionClass { + v.RetentionClass = f.Text + break + } + } + } + out.Events = append(out.Events, v) + } + if err = rows.Err(); err != nil { + return out, err + } + if len(out.Events) > req.Limit { + out.Events = out.Events[:req.Limit] + out.HasMore = true + } + if len(out.Events) > 0 { + out.NextCursor = out.Events[len(out.Events)-1].Cursor + } + return out, nil +} + +func (s *Store) DeleteRetentionEvents(ctx context.Context, req retention.DeleteRequest) (retention.DeleteResult, error) { + var out retention.DeleteResult + if len(req.Events) == 0 { + return out, nil + } + if req.DryRun { + return out, nil + } + if req.ArchiveID == "" { + return out, fmt.Errorf("trail/postgres: archive receipt is required") + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return out, err + } + defer tx.Rollback(ctx) + seen := map[string]bool{} + for _, e := range req.Events { + for _, p := range e.Payloads { + k := p.Ref.Store + "\x00" + p.Ref.Key + at, ok := req.PayloadEligibleAt[k] + if !ok || seen[k] || p.Ref.Store == "" || p.Ref.Key == "" { + continue + } + seen[k] = true + b, er := json.Marshal(p.Ref) + if er != nil { + return out, er + } + _, er = tx.Exec(ctx, `INSERT INTO trail_payload_gc_candidates(store,object_key,ref,eligible_at,archive_id) VALUES($1,$2,$3,$4,$5) ON CONFLICT(store,object_key) DO UPDATE SET eligible_at=GREATEST(trail_payload_gc_candidates.eligible_at,EXCLUDED.eligible_at),ref=EXCLUDED.ref,archive_id=EXCLUDED.archive_id`, p.Ref.Store, p.Ref.Key, b, at, req.ArchiveID) + if er != nil { + return out, er + } + out.PayloadCandidates++ + } + } + ids := make([][]byte, 0, len(req.Events)) + for _, e := range req.Events { + id, er := hex.DecodeString(e.ID) + if er != nil { + return out, er + } + ids = append(ids, id) + } + var archived int + if err = tx.QueryRow(ctx, `SELECT count(*) FROM trail_event_archives WHERE archive_id=$1 AND event_id=ANY($2)`, req.ArchiveID, ids).Scan(&archived); err != nil { + return out, err + } + if archived != len(ids) { + return out, fmt.Errorf("trail/postgres: archive receipt does not cover deletion batch") + } + ct, err := tx.Exec(ctx, `DELETE FROM trail_events WHERE event_id=ANY($1)`, ids) + if err != nil { + return out, err + } + out.EventsDeleted = uint64(ct.RowsAffected()) + if err = tx.Commit(ctx); err != nil { + return retention.DeleteResult{}, err + } + return out, nil +} + +func (s *Store) LoadArchiveEvents(ctx context.Context, events []retention.Event) ([]storage.EventRecord, error) { + ids := make([][]byte, 0, len(events)) + for _, e := range events { + id, err := hex.DecodeString(e.ID) + if err != nil { + return nil, err + } + ids = append(ids, id) + } + rows, err := s.pool.Query(ctx, eventSelect+` WHERE event_id=ANY($1) ORDER BY event_time,event_id`, ids) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]storage.EventRecord, 0, len(events)) + for rows.Next() { + v, err := scanEvent(rows) + if err != nil { + return nil, err + } + out = append(out, v) + } + return out, rows.Err() +} +func (s *Store) RecordArchive(ctx context.Context, receipt archive.Receipt) error { + manifest, err := json.Marshal(receipt.Manifest) + if err != nil { + return err + } + ref, err := json.Marshal(receipt.ManifestRef) + if err != nil { + return err + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + if _, err = tx.Exec(ctx, `INSERT INTO trail_archives(archive_id,manifest,manifest_ref,created_at) VALUES($1,$2,$3,$4) ON CONFLICT(archive_id) DO NOTHING`, receipt.Manifest.ID, manifest, ref, receipt.Manifest.CreatedAt); err != nil { + return err + } + for _, id := range receipt.Manifest.EventIDs { + parsed, er := trail.ParseEventID(id) + if er != nil { + return er + } + raw := parsed[:] + if _, er = tx.Exec(ctx, `INSERT INTO trail_event_archives(event_id,archive_id) VALUES($1,$2) ON CONFLICT(event_id) DO UPDATE SET archive_id=EXCLUDED.archive_id`, raw, receipt.Manifest.ID); er != nil { + return er + } + } + return tx.Commit(ctx) +} + +func (s *Store) ListArchives(ctx context.Context, offset, limit int) (archive.Page, error) { + if offset < 0 || limit < 1 || limit > 500 { + return archive.Page{}, storage.ErrInvalidCursor + } + rows, err := s.pool.Query(ctx, `SELECT manifest,manifest_ref FROM trail_archives ORDER BY created_at DESC,archive_id LIMIT $1 OFFSET $2`, limit+1, offset) + if err != nil { + return archive.Page{}, err + } + defer rows.Close() + var out archive.Page + for rows.Next() { + var manifestJSON, refJSON []byte + if err = rows.Scan(&manifestJSON, &refJSON); err != nil { + return out, err + } + var receipt archive.Receipt + if err = json.Unmarshal(manifestJSON, &receipt.Manifest); err != nil { + return out, err + } + if err = json.Unmarshal(refJSON, &receipt.ManifestRef); err != nil { + return out, err + } + out.Items = append(out.Items, receipt) + } + if len(out.Items) > limit { + out.Items = out.Items[:limit] + out.HasMore = true + } + return out, rows.Err() +} + +func (s *Store) GetArchive(ctx context.Context, id string) (archive.Receipt, error) { + var manifestJSON, refJSON []byte + if err := s.pool.QueryRow(ctx, `SELECT manifest,manifest_ref FROM trail_archives WHERE archive_id=$1`, id).Scan(&manifestJSON, &refJSON); err != nil { + if strings.Contains(err.Error(), "no rows") { + return archive.Receipt{}, storage.ErrNotFound + } + return archive.Receipt{}, err + } + var out archive.Receipt + if err := json.Unmarshal(manifestJSON, &out.Manifest); err != nil { + return out, err + } + if err := json.Unmarshal(refJSON, &out.ManifestRef); err != nil { + return out, err + } + return out, nil +} + +// RestoreArchiveEvents performs an idempotent merge using the event_id primary +// key. Existing live events are intentionally preserved. +func (s *Store) RestoreArchiveEvents(ctx context.Context, _ string, events []storage.EventRecord) (int, error) { + if len(events) == 0 { + return 0, nil + } + if err := s.WriteBatches(ctx, []storage.Batch{{Events: events}}); err != nil { + return 0, err + } + return len(events), nil +} + +func (s *Store) ListPayloadCandidates(ctx context.Context, before time.Time, limit int) (retention.CandidatePage, error) { + rows, err := s.pool.Query(ctx, `SELECT ref,archive_id,eligible_at,candidate_at,attempts,last_error FROM trail_payload_gc_candidates WHERE archive_id<>'' AND eligible_at<=now() AND candidate_at<=$1 AND retry_after<=now() ORDER BY candidate_at,store,object_key LIMIT $2`, before, limit) + if err != nil { + return retention.CandidatePage{}, err + } + defer rows.Close() + var out retention.CandidatePage + for rows.Next() { + var raw []byte + var c retention.Candidate + if err = rows.Scan(&raw, &c.ArchiveID, &c.EligibleAt, &c.CandidateAt, &c.Attempts, &c.LastError); err != nil { + return out, err + } + if err = json.Unmarshal(raw, &c.Ref); err != nil { + return out, err + } + out.Candidates = append(out.Candidates, c) + } + return out, rows.Err() +} +func (s *Store) ReferencedPayloads(ctx context.Context, refs []payload.Ref) (map[string]bool, error) { + out := map[string]bool{} + for _, ref := range refs { + var yes bool + err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM trail_events e CROSS JOIN LATERAL jsonb_array_elements(e.payload_refs) p WHERE p->'Ref'->>'store'=$1 AND p->'Ref'->>'key'=$2 LIMIT 1)`, ref.Store, ref.Key).Scan(&yes) + if err != nil { + return nil, err + } + out[ref.Store+"\x00"+ref.Key] = yes + } + return out, nil +} +func (s *Store) CompletePayloadCandidate(ctx context.Context, ref payload.Ref, failure string) error { + if failure == "" || failure == "referenced" { + _, err := s.pool.Exec(ctx, `DELETE FROM trail_payload_gc_candidates WHERE store=$1 AND object_key=$2`, ref.Store, ref.Key) + return err + } + _, err := s.pool.Exec(ctx, `UPDATE trail_payload_gc_candidates SET attempts=attempts+1,last_error=$3,retry_after=now()+interval '1 hour' WHERE store=$1 AND object_key=$2`, ref.Store, ref.Key, failure) + return err +} + +func retentionCursor(t time.Time, id []byte) string { + return strconv.FormatInt(t.UnixNano(), 10) + ":" + hex.EncodeToString(id) +} +func parseRetentionCursor(v string) (time.Time, []byte, error) { + a, b, ok := strings.Cut(v, ":") + if !ok { + return time.Time{}, nil, fmt.Errorf("invalid retention cursor") + } + n, err := strconv.ParseInt(a, 10, 64) + if err != nil { + return time.Time{}, nil, err + } + id, err := hex.DecodeString(b) + return time.Unix(0, n).UTC(), id, err +} +func retentionLevel(v string) trail.Level { + switch v { + case "debug": + return trail.LevelDebug + case "warn": + return trail.LevelWarn + case "error": + return trail.LevelError + default: + return trail.LevelInfo + } +} + +var _ retention.Store = (*Store)(nil) +var _ archive.Catalog = (*Store)(nil) +var _ archive.RestoreStore = (*Store)(nil) + +func (s *Store) ReconcileRetention(context.Context) error { return nil } diff --git a/storage/postgres/retention_test.go b/storage/postgres/retention_test.go new file mode 100644 index 0000000..0f7babf --- /dev/null +++ b/storage/postgres/retention_test.go @@ -0,0 +1,36 @@ +package postgres + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/vestavision/trail/retention" +) + +func TestRetentionCursorRoundTrip(t *testing.T) { + at := time.Unix(123, 456).UTC() + id := bytes.Repeat([]byte{7}, 16) + cursor := retentionCursor(at, id) + gotAt, gotID, err := parseRetentionCursor(cursor) + if err != nil { + t.Fatal(err) + } + if !gotAt.Equal(at) || !bytes.Equal(gotID, id) { + t.Fatalf("at=%v id=%x", gotAt, gotID) + } +} + +func TestDeleteRequiresArchiveReceiptBeforeDatabaseAccess(t *testing.T) { + store := &Store{} + _, err := store.DeleteRetentionEvents(context.Background(), retention.DeleteRequest{Events: []retention.Event{{ID: "00"}}}) + if err == nil { + t.Fatal("expected archive receipt error") + } +} +func TestRetentionCursorRejectsInvalid(t *testing.T) { + if _, _, err := parseRetentionCursor("invalid"); err == nil { + t.Fatal("expected invalid cursor error") + } +} diff --git a/storage/storage.go b/storage/storage.go index 5e39c4b..1da3af4 100644 --- a/storage/storage.go +++ b/storage/storage.go @@ -75,6 +75,7 @@ type EventRecord struct { FlowStatus string ExecutionStatus string ExecutionKind string + RetentionClass string HTTP HTTPRecord Payloads []PayloadLink BatchID trail.EventID