From 4923a87a736b1d34e0681ecb4bdab03e5dd51150 Mon Sep 17 00:00:00 2001 From: Soham Shirke <129677811+SShogun@users.noreply.github.com> Date: Tue, 15 Sep 2026 21:08:12 +0530 Subject: [PATCH] hardening: make audit producer failures observable Centralize best-effort audit emission with structured failure logging and Prometheus producer metrics. Emit create, update, and delete audit events consistently, add regression coverage, and document the exact failure semantics for producer appends, stream recovery, cache degradation, Bloom idempotency, and search. --- README.md | 22 +++-- docs/README.md | 23 +++-- docs/failure-semantics.md | 115 +++++++++++++++++++++++ internal/app/app.go | 8 +- internal/audit/emitter.go | 83 +++++++++++++++++ internal/audit/emitter_test.go | 64 +++++++++++++ internal/handlers/audit_events_test.go | 123 +++++++++++++++++++++++++ internal/handlers/items_create.go | 27 ++---- internal/handlers/items_create_test.go | 6 +- internal/handlers/items_delete.go | 12 ++- internal/handlers/items_update.go | 27 ++---- internal/observability/metrics.go | 30 ++++++ 12 files changed, 480 insertions(+), 60 deletions(-) create mode 100644 docs/failure-semantics.md create mode 100644 internal/audit/emitter.go create mode 100644 internal/audit/emitter_test.go create mode 100644 internal/handlers/audit_events_test.go diff --git a/README.md b/README.md index 068ed5e..611fa64 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ If Redis feels messy, this repo is meant to be the place you reopen and revise f | Project tour | [docs/README.md](docs/README.md) | | How the service is wired | [docs/implementation/architecture.md](docs/implementation/architecture.md) | | Redis module choices | [docs/implementation/redis-patterns.md](docs/implementation/redis-patterns.md) | +| Failure and delivery semantics | [docs/failure-semantics.md](docs/failure-semantics.md) | | Phase-by-phase build history | [docs/REDISFORGE_BUILD_GUIDE.md](docs/REDISFORGE_BUILD_GUIDE.md) | | Redis configuration tradeoffs | [docs/redis-decisions.md](docs/redis-decisions.md) | | Profiling and tuning | [docs/profiling-results.md](docs/profiling-results.md) | @@ -50,7 +51,8 @@ graph TD subgraph Redis Modules Handlers -- "Idempotency Check" --> RedisBloom[(RedisBloom)] - Handlers -- "Emit Audit Event" --> Streams[(Redis Streams)] + Handlers -- "Best-effort Audit Event" --> AuditEmitter[Audit Emitter] + AuditEmitter -- "XADD" --> Streams[(Redis Streams)] Handlers -- "Search Query" --> RediSearch[(RediSearch)] RediSearch -. "Indexes" .-> RedisJSON end @@ -58,7 +60,8 @@ graph TD Streams --> Worker[Background Audit Worker] subgraph Observability - API -.-> Metrics["/healthz and /metrics"] + AuditEmitter -.-> Metrics["/healthz and /metrics"] + API -.-> Metrics API -.-> OTel[OpenTelemetry Tracing] end ``` @@ -72,7 +75,7 @@ The important design choice: the domain stays tiny so Redis remains the main thi | RedisJSON | Store Item documents and support partial updates | | RedisBloom | Idempotency pre-checks with no false negatives | | RediSearch | Full-text search, category filters, tag filters, score ranges | -| Streams | Audit-event processing with consumer groups and at-least-once stale-message recovery after append | +| Streams | Best-effort producer audit events plus at-least-once stale-message recovery after successful append | | Pub/Sub | Ephemeral real-time notifications | | Sentinel | High-availability topology support | | Cluster client | Horizontal-scale topology support and hash-tag discipline | @@ -85,6 +88,7 @@ redisforge/ |-- cmd/redisforge/ # application entrypoint |-- internal/ | |-- app/ # dependency wiring and lifecycle +| |-- audit/ # audit producer, timeout, logging, metrics | |-- config/ # typed environment configuration | |-- domain/ # Item model and sentinel errors | |-- handlers/ # HTTP handlers for CRUD and search @@ -102,6 +106,7 @@ redisforge/ |-- docs/ | |-- implementation/ # architecture and Redis pattern notes | |-- README.md # docs index and learning path +| |-- failure-semantics.md # explicit degraded-mode and delivery guarantees | |-- REDISFORGE_BUILD_GUIDE.md | |-- redis-decisions.md | |-- profiling-results.md @@ -173,11 +178,11 @@ make down | --- | --- | --- | | GET | `/healthz` | Health check | | GET | `/metrics` | Prometheus metrics | -| POST | `/v1/items` | Create an item and emit an audit event | +| POST | `/v1/items` | Create an item and schedule a best-effort `created` audit event | | GET | `/v1/items` | List items | | GET | `/v1/items/{id}` | Fetch one item through cache-aside lookup | -| PUT | `/v1/items/{id}` | Update item and refresh Redis state | -| DELETE | `/v1/items/{id}` | Delete item | +| PUT | `/v1/items/{id}` | Update an item and schedule a best-effort `updated` audit event | +| DELETE | `/v1/items/{id}` | Delete an item and schedule a best-effort `deleted` audit event | | GET | `/v1/items/search?q=...` | Search through RediSearch | ## Tests @@ -186,7 +191,7 @@ make down .\make.ps1 test ``` -The tests use `testcontainers-go` where Redis behavior matters. Handler integration tests run against a pinned Redis Stack image, while Streams recovery tests run against a pinned Redis image and prove group bootstrap replay, cursor-complete stale claiming, and ACK-failure handling. +The tests use `testcontainers-go` where Redis behavior matters. Handler integration tests run against a pinned Redis Stack image, while Streams recovery tests run against a pinned Redis image and prove group bootstrap replay, cursor-complete stale claiming, and ACK-failure handling. Producer tests also prove create/update/delete action emission and synchronous append error propagation. ## Development Rhythm @@ -215,11 +220,12 @@ RedisForge currently implements the planned learning phases from bootstrap throu - Redis client abstraction for single-node, Sentinel, and Cluster modes - RedisJSON, RedisBloom, RediSearch, Streams, and Pub/Sub wrappers - Cache-aside repository pattern +- Best-effort audit producer for create/update/delete with structured failure logs and Prometheus metrics - Audit stream worker with at-least-once recovery for successfully appended events - Prometheus metrics and OpenTelemetry hooks - Integration tests and benchmark/demo scripts -The item handlers currently emit audit events asynchronously on a best-effort basis. A failed `XADD` does not roll back the item write. See [Redis configuration decisions](docs/redis-decisions.md) for the exact delivery and recovery contract. +The item handlers intentionally return independently of audit `XADD`. A failed producer append does not roll back the item write, but it is now visible through logs plus `audit_events_emitted_total{action,status}` and `audit_emit_latency_ms{action,status}`. Once an event is successfully appended, worker recovery is at least once and duplicates remain possible. See [Failure Semantics](docs/failure-semantics.md) for the exact contract. ## Repository Goal diff --git a/docs/README.md b/docs/README.md index 7011527..e6677f2 100644 --- a/docs/README.md +++ b/docs/README.md @@ -16,16 +16,19 @@ This folder is the RedisForge learning path. Use it when you want to revise Redi 4. [Redis Decisions](redis-decisions.md) Deeper notes on configuration choices, recovery guarantees, and tradeoffs. -5. [Profiling Results](profiling-results.md) and [Hot-Path Benchmarks](hotpath-benchmarks.md) +5. [Failure Semantics](failure-semantics.md) + Exact behavior when audit emission, cache operations, Bloom checks, stream recovery, or search operations fail. + +6. [Profiling Results](profiling-results.md) and [Hot-Path Benchmarks](hotpath-benchmarks.md) Use the profiling report for the historical load-test snapshot and the benchmark guide for the current reproducible Redis hot-path harness. -6. [Demo Workflow](demo_workflow.md) +7. [Demo Workflow](demo_workflow.md) A short repeatable flow for recording demos and proving the project works. -7. [Project Journal](project-journal.md) +8. [Project Journal](project-journal.md) Commit cadence, next public updates, and the habit loop that keeps the repo looking alive. -8. [Interview Questions](interview_questions.md) and [Interview Answers](redisforge_interview_answers.md) +9. [Interview Questions](interview_questions.md) and [Interview Answers](redisforge_interview_answers.md) 52 backend interview questions with RedisForge-specific answers. Use for revision. ## Docs By Intent @@ -34,6 +37,7 @@ This folder is the RedisForge learning path. Use it when you want to revise Redi | --- | --- | | "I forgot how Redis Streams consumer groups work." | [implementation/redis-patterns.md](implementation/redis-patterns.md) + [redis-decisions.md](redis-decisions.md) | | "I want to understand the codebase fast." | [implementation/architecture.md](implementation/architecture.md) | +| "What happens when Redis or audit emission fails?" | [failure-semantics.md](failure-semantics.md) | | "I want to know why these Redis settings exist." | [redis-decisions.md](redis-decisions.md) | | "I want to benchmark Redis hot paths reproducibly." | [hotpath-benchmarks.md](hotpath-benchmarks.md) | | "I want to record a demo for GitHub/LinkedIn/X." | [demo_workflow.md](demo_workflow.md) and [social_templates.md](social_templates.md) | @@ -48,18 +52,19 @@ Every meaningful feature should leave behind three things: **code**, a **test**, | Feature | Code | Test | Doc Note | | --- | --- | --- | --- | | RedisJSON (CRUD + partial updates) | `redisx/json.go` | `redisx/json_test.go` ✅ | `redis-patterns.md` ✅ | -| RedisBloom (idempotency) | `redisx/bloom.go` | `redisx/bloom_test.go` + `handlers/items_create_test.go` ✅ | `redis-patterns.md` ✅ | +| RedisBloom (idempotency) | `redisx/bloom.go` | `redisx/bloom_test.go` + `handlers/items_create_test.go` ✅ | `redis-patterns.md` + `failure-semantics.md` ✅ | | RediSearch (full-text + faceted) | `redisx/search.go` | `redisx/search_test.go` ✅ | `redis-patterns.md` ✅ | -| Redis Streams (audit log + stale recovery) | `redisx/streams.go` + `workers/audit_workers.go` | `workers/audit_workers_test.go` ✅ | `redis-decisions.md` ✅ | +| Audit producer (create/update/delete) | `audit/emitter.go` + item handlers | `audit/emitter_test.go` + `handlers/audit_events_test.go` ✅ | `failure-semantics.md` ✅ | +| Redis Streams (audit log + stale recovery) | `redisx/streams.go` + `workers/audit_workers.go` | `workers/audit_workers_test.go` ✅ | `redis-decisions.md` + `failure-semantics.md` ✅ | | Pub/Sub (ephemeral notifications) | `redisx/pubsub.go` | ⚠️ needs dedicated test | `redis-patterns.md` ✅ | -| Cache-Aside Repository | `repo/item_cache.go` | `repo/item_cache_test.go` (4 tests + 4 benchmarks) | `redis-patterns.md` ✅ | +| Cache-Aside Repository | `repo/item_cache.go` | `repo/item_cache_test.go` (4 tests + 4 benchmarks) | `redis-patterns.md` + `failure-semantics.md` ✅ | | Redis hot paths | JSON/Bloom/Search/Streams wrappers | `redisx/hotpaths_benchmark_test.go` + CI smoke ✅ | `hotpath-benchmarks.md` ✅ | | Sentinel HA | `redisx/client.go` | Manual via `redis-sentinel/` compose | `redis-decisions.md` ✅ | | Cluster (horizontal scale) | `redisx/client.go` | Manual via `redis-cluster/` compose | `redis-cluster/README.md` ✅ | -| Prometheus Metrics | `observability/metrics.go` | Verified via `/metrics` endpoint | `profiling-results.md` ✅ | +| Prometheus Metrics | `observability/metrics.go` | Verified via `/metrics` endpoint | `failure-semantics.md` + `profiling-results.md` ✅ | | OpenTelemetry Tracing | `observability/tracing_hooks.go` | Verified via trace output | `architecture.md` ✅ | | Graceful Shutdown | `app/app.go` | Manual verification | `architecture.md` ✅ | -| Handler Idempotency (E2E) | `handlers/items_create.go` | `handlers/items_create_test.go` ✅ | `redis-patterns.md` ✅ | +| Handler Idempotency (E2E) | `handlers/items_create.go` | `handlers/items_create_test.go` ✅ | `redis-patterns.md` + `failure-semantics.md` ✅ | > **Legend:** ✅ = covered, ⚠️ = gap identified for future commit diff --git a/docs/failure-semantics.md b/docs/failure-semantics.md new file mode 100644 index 0000000..b06a195 --- /dev/null +++ b/docs/failure-semantics.md @@ -0,0 +1,115 @@ +# RedisForge Failure Semantics + +This document states what RedisForge guarantees when Redis operations fail. It is intentionally narrower than a production outbox or transactional-event design: the project keeps item writes and Redis audit emission separate so the trade-offs remain visible. + +## Audit event lifecycle + +For successful item creates, updates, and deletes, the handler schedules a best-effort audit event with action `created`, `updated`, or `deleted`. + +```text +HTTP write succeeds + | + +--> response can return immediately + | + `--> async audit emitter + | + +--> JSON serialize + +--> XADD audit-events (5s timeout) + | + `--> audit worker consumer group + | + +--> process event + `--> XACK +``` + +### Before the event reaches Redis + +The producer path is asynchronous and best effort. + +- A successful item write is **not rolled back** when audit `XADD` fails. +- Producer append failures are logged with `event_id`, `item_id`, `action`, and the error. +- `audit_events_emitted_total{action,status}` counts producer append outcomes. +- `audit_emit_latency_ms{action,status}` measures serialization plus append latency. +- Each async append has a 5-second timeout. +- There is currently no producer retry queue, transactional outbox, or disk-backed handoff. +- Because producer goroutines are not drained during shutdown, an in-flight best-effort event can also be lost if the process exits before the append finishes. + +Operationally, any sustained increase in: + +```promql +rate(audit_events_emitted_total{status="error"}[5m]) +``` + +means item writes may be succeeding without corresponding audit entries. + +### After the event reaches Redis + +Once `XADD` succeeds, the worker contract is **at least once**, not exactly once. + +- The consumer reads new group entries with `XREADGROUP`. +- Delivered but unacknowledged entries remain in the Pending Entries List (PEL). +- Stale pending entries are recovered with cursor-complete `XAUTOCLAIM` scanning. +- The worker ACKs only after its processing step succeeds. +- ACK failures are treated as processing failures, so the entry remains recoverable. +- A crash after processing but before ACK can cause the same event to be delivered again. + +Any future worker side effect must therefore be idempotent or explicitly tolerate duplicate delivery. + +### Malformed audit entries + +Malformed entries are not useful work and are ACKed to prevent a permanent poison-message loop. If that cleanup ACK fails, the failure is surfaced instead of being reported as success, so the pending entry can be reclaimed again. + +RedisForge does not currently maintain a dead-letter stream containing malformed payloads. The current policy is **log + ACK/drop**, with ACK failure remaining visible. + +## Item cache failures + +`CacheItemRepo` treats RedisJSON as a cache in front of the fallback repository. + +| Operation | Redis/cache failure behavior | Item operation result | +| --- | --- | --- | +| Create | Fallback create succeeds, cache write-through failure is logged | Create still succeeds | +| Get | Cache miss or cache read error falls through to fallback; successful fallback read attempts cache backfill | Read can still succeed | +| Update | Fallback update succeeds, cache invalidation failure is logged | Update still succeeds | +| Delete | Fallback delete succeeds; cache deletion is best effort | Delete still succeeds | + +This is deliberate cache degradation behavior: Redis cache availability does not define item durability in the current architecture. + +## Bloom-filter idempotency failures + +The create handler uses the Bloom filter as a demonstration pre-check rather than a durable idempotency store. + +- If `BF.EXISTS` returns `true`, the request is rejected as a duplicate. +- If the Bloom lookup itself errors, the handler currently **fails open** and continues the create path. +- After a successful create, `BF.ADD` is best effort; its error does not fail the HTTP request. +- A lost Bloom update can therefore allow a later request with the same idempotency key to create another item. + +The Bloom filter must not be interpreted as a transactional exactly-once request guarantee. + +## Search failures + +RediSearch is on the request path for the search endpoint. An `FT.SEARCH` error is returned to the handler and surfaces as a failed request; there is no fallback full scan. + +## What RedisForge does not claim + +RedisForge does **not** currently claim: + +- atomic item-write + audit-event commit +- exactly-once audit delivery or processing +- durable producer retries before `XADD` +- transactional idempotency based on the Bloom filter +- cache writes as the source of truth + +Those would require different architecture, such as a transactional outbox, durable retry queue, or a database-backed idempotency record. + +## Verification map + +| Failure semantic | Proof | +| --- | --- | +| create/update/delete schedule the correct audit action | `internal/handlers/audit_events_test.go` | +| producer append errors are returned by the synchronous emitter path | `internal/audit/emitter_test.go` | +| pre-existing stream entries replay after group creation | `internal/workers/audit_workers_test.go` | +| stale PEL recovery traverses the `XAUTOCLAIM` cursor | `internal/workers/audit_workers_test.go` | +| ACK failures remain processing failures | `internal/workers/audit_workers_test.go` | +| real Redis module integration | Testcontainers coverage under `internal/redisx` and `internal/handlers` | + +The `/metrics` endpoint is the operational proof surface for producer failures, worker processing, pending counts, RedisJSON latency, Bloom checks, and search latency. diff --git a/internal/app/app.go b/internal/app/app.go index 5c38c51..94b2853 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -19,6 +19,7 @@ import ( "syscall" "time" + "github.com/SShogun/redisforge/internal/audit" "github.com/SShogun/redisforge/internal/config" "github.com/SShogun/redisforge/internal/handlers" "github.com/SShogun/redisforge/internal/logging" @@ -59,6 +60,7 @@ func Run() error { streamStore := redisx.NewStreamClient(redisClient) pubSubStore := redisx.NewPubSubClient(redisClient) _ = pubSubStore // used in handlers optionally + auditEmitter := audit.NewEmitter(streamStore, logger) // Initialise bloom filter (idempotent) if err := bloomStore.Reserve(ctx, 0.001, 1_000_000); err != nil { @@ -104,12 +106,12 @@ func Run() error { r.Handle("/metrics", promhttp.Handler()) r.Route("/v1/items", func(r chi.Router) { - r.Post("/", handlers.HandleCreateItem(cacheRepo, streamStore, bloomStore)) + r.Post("/", handlers.HandleCreateItem(cacheRepo, auditEmitter, bloomStore)) r.Get("/search", handlers.HandleSearchItems(searchStore)) r.Get("/", handlers.HandleListItems(cacheRepo)) r.Get("/{id}", handlers.HandleGetItem(cacheRepo)) - r.Put("/{id}", handlers.HandleUpdateItem(cacheRepo, streamStore)) - r.Delete("/{id}", handlers.HandleDeleteItem(cacheRepo)) + r.Put("/{id}", handlers.HandleUpdateItem(cacheRepo, auditEmitter)) + r.Delete("/{id}", handlers.HandleDeleteItem(cacheRepo, auditEmitter)) }) // ── HTTP Server ──────────────────────────────────────────────── diff --git a/internal/audit/emitter.go b/internal/audit/emitter.go new file mode 100644 index 0000000..8d8b7f0 --- /dev/null +++ b/internal/audit/emitter.go @@ -0,0 +1,83 @@ +package audit + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/SShogun/redisforge/internal/domain" + "github.com/SShogun/redisforge/internal/observability" +) + +const ( + StreamName = "audit-events" + emitTimeout = 5 * time.Second +) + +type StreamAppender interface { + Append(context.Context, string, map[string]interface{}) (string, error) +} + +// Emitter owns producer-side audit serialization, timeout, metrics, and failure logging. +// HTTP handlers intentionally use EmitAsync so item writes are not rolled back when Redis +// is unavailable. Once an event is appended, the audit worker provides at-least-once recovery. +type Emitter struct { + stream StreamAppender + logger *slog.Logger + timeout time.Duration +} + +func NewEmitter(stream StreamAppender, logger *slog.Logger) *Emitter { + if logger == nil { + logger = slog.Default() + } + return &Emitter{stream: stream, logger: logger, timeout: emitTimeout} +} + +// Emit appends one audit event synchronously and returns any serialization or Redis error. +func (e *Emitter) Emit(ctx context.Context, event domain.AuditEvent) (err error) { + start := time.Now() + defer func() { + observability.RecordAuditEmit(start, event.Action, err) + }() + + if e == nil || e.stream == nil { + return fmt.Errorf("audit.Emitter.Emit: stream is nil") + } + + payload, err := json.Marshal(event) + if err != nil { + return fmt.Errorf("audit.Emitter.Emit: marshal: %w", err) + } + + if _, err := e.stream.Append(ctx, StreamName, map[string]interface{}{ + "event": string(payload), + }); err != nil { + return fmt.Errorf("audit.Emitter.Emit: append: %w", err) + } + return nil +} + +// EmitAsync preserves the current best-effort producer contract: the HTTP response is not +// blocked on Redis. Failures are still visible through structured logs and Prometheus metrics. +func (e *Emitter) EmitAsync(event domain.AuditEvent) { + if e == nil { + return + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), e.timeout) + defer cancel() + + if err := e.Emit(ctx, event); err != nil { + e.logger.Error( + "audit event emission failed", + "event_id", event.EventID, + "item_id", event.ItemID, + "action", event.Action, + "err", err, + ) + } + }() +} diff --git a/internal/audit/emitter_test.go b/internal/audit/emitter_test.go new file mode 100644 index 0000000..cebbe3e --- /dev/null +++ b/internal/audit/emitter_test.go @@ -0,0 +1,64 @@ +package audit + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + + "github.com/SShogun/redisforge/internal/domain" +) + +type fakeAppender struct { + stream string + fields map[string]interface{} + err error +} + +func (f *fakeAppender) Append(_ context.Context, stream string, fields map[string]interface{}) (string, error) { + f.stream = stream + f.fields = fields + if f.err != nil { + return "", f.err + } + return "1-0", nil +} + +func TestEmitterEmit(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + t.Run("appends serialized event to audit stream", func(t *testing.T) { + appender := &fakeAppender{} + emitter := NewEmitter(appender, logger) + event := domain.AuditEvent{EventID: "event-1", ItemID: "item-1", Action: "created"} + + if err := emitter.Emit(context.Background(), event); err != nil { + t.Fatalf("Emit: %v", err) + } + if appender.stream != StreamName { + t.Fatalf("expected stream %q, got %q", StreamName, appender.stream) + } + raw, ok := appender.fields["event"].(string) + if !ok || raw == "" { + t.Fatalf("expected serialized event field, got %#v", appender.fields["event"]) + } + }) + + t.Run("propagates append failure", func(t *testing.T) { + want := errors.New("redis unavailable") + emitter := NewEmitter(&fakeAppender{err: want}, logger) + + err := emitter.Emit(context.Background(), domain.AuditEvent{Action: "updated"}) + if !errors.Is(err, want) { + t.Fatalf("expected wrapped append error, got %v", err) + } + }) + + t.Run("rejects nil stream", func(t *testing.T) { + emitter := NewEmitter(nil, logger) + if err := emitter.Emit(context.Background(), domain.AuditEvent{Action: "deleted"}); err == nil { + t.Fatal("expected nil stream error") + } + }) +} diff --git a/internal/handlers/audit_events_test.go b/internal/handlers/audit_events_test.go new file mode 100644 index 0000000..08de22c --- /dev/null +++ b/internal/handlers/audit_events_test.go @@ -0,0 +1,123 @@ +package handlers_test + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/SShogun/redisforge/internal/audit" + "github.com/SShogun/redisforge/internal/domain" + "github.com/SShogun/redisforge/internal/handlers" + "github.com/SShogun/redisforge/internal/repo" + "github.com/go-chi/chi/v5" +) + +type recordingAuditAppender struct { + events chan domain.AuditEvent +} + +func (r *recordingAuditAppender) Append(_ context.Context, stream string, fields map[string]interface{}) (string, error) { + if stream != audit.StreamName { + return "", nil + } + raw, _ := fields["event"].(string) + var event domain.AuditEvent + if err := json.Unmarshal([]byte(raw), &event); err == nil { + r.events <- event + } + return "1-0", nil +} + +func TestItemWriteHandlersEmitAuditEvents(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + t.Run("create", func(t *testing.T) { + items := repo.NewMemoryItemRepo() + appender := &recordingAuditAppender{events: make(chan domain.AuditEvent, 1)} + emitter := audit.NewEmitter(appender, logger) + handler := handlers.HandleCreateItem(items, emitter, nil) + + req := httptest.NewRequest(http.MethodPost, "/v1/items", bytes.NewBufferString(`{"name":"created","category":"tests"}`)) + req.Header.Set("Content-Type", "application/json") + rr := httptest.NewRecorder() + handler.ServeHTTP(rr, req) + if rr.Code != http.StatusCreated { + t.Fatalf("expected %d, got %d", http.StatusCreated, rr.Code) + } + + event := waitForAuditEvent(t, appender.events) + if event.Action != "created" || event.ItemID == "" { + t.Fatalf("unexpected create audit event: %#v", event) + } + }) + + t.Run("update", func(t *testing.T) { + items := repo.NewMemoryItemRepo() + created, err := items.Create(context.Background(), domain.Item{ID: "item-update", Name: "before", Category: "tests"}) + if err != nil { + t.Fatalf("seed item: %v", err) + } + if created.ID == "" { + t.Fatal("seed item missing id") + } + + appender := &recordingAuditAppender{events: make(chan domain.AuditEvent, 1)} + emitter := audit.NewEmitter(appender, logger) + router := chi.NewRouter() + router.Put("/v1/items/{id}", handlers.HandleUpdateItem(items, emitter)) + + req := httptest.NewRequest(http.MethodPut, "/v1/items/item-update", bytes.NewBufferString(`{"name":"after"}`)) + req.Header.Set("Content-Type", "application/json") + rr := httptest.NewRecorder() + router.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("expected %d, got %d", http.StatusOK, rr.Code) + } + + event := waitForAuditEvent(t, appender.events) + if event.Action != "updated" || event.ItemID != "item-update" { + t.Fatalf("unexpected update audit event: %#v", event) + } + }) + + t.Run("delete", func(t *testing.T) { + items := repo.NewMemoryItemRepo() + if _, err := items.Create(context.Background(), domain.Item{ID: "item-delete", Name: "delete", Category: "tests"}); err != nil { + t.Fatalf("seed item: %v", err) + } + + appender := &recordingAuditAppender{events: make(chan domain.AuditEvent, 1)} + emitter := audit.NewEmitter(appender, logger) + router := chi.NewRouter() + router.Delete("/v1/items/{id}", handlers.HandleDeleteItem(items, emitter)) + + req := httptest.NewRequest(http.MethodDelete, "/v1/items/item-delete", nil) + rr := httptest.NewRecorder() + router.ServeHTTP(rr, req) + if rr.Code != http.StatusNoContent { + t.Fatalf("expected %d, got %d", http.StatusNoContent, rr.Code) + } + + event := waitForAuditEvent(t, appender.events) + if event.Action != "deleted" || event.ItemID != "item-delete" { + t.Fatalf("unexpected delete audit event: %#v", event) + } + }) +} + +func waitForAuditEvent(t *testing.T, events <-chan domain.AuditEvent) domain.AuditEvent { + t.Helper() + select { + case event := <-events: + return event + case <-time.After(time.Second): + t.Fatal("timed out waiting for audit event") + return domain.AuditEvent{} + } +} diff --git a/internal/handlers/items_create.go b/internal/handlers/items_create.go index 37c709c..836482c 100644 --- a/internal/handlers/items_create.go +++ b/internal/handlers/items_create.go @@ -1,12 +1,11 @@ package handlers import ( - "context" - "encoding/json" "net/http" "strings" "time" + "github.com/SShogun/redisforge/internal/audit" "github.com/SShogun/redisforge/internal/domain" "github.com/SShogun/redisforge/internal/redisx" "github.com/SShogun/redisforge/internal/repo" @@ -17,7 +16,7 @@ import ( // Idempotency pre-check: BloomFilter → if "might exist", skip to duplicate check. func HandleCreateItem( items repo.ItemRepo, - stream *redisx.StreamClient, + auditEmitter *audit.Emitter, bloom *redisx.BloomFilter, ) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { @@ -67,22 +66,12 @@ func HandleCreateItem( _ = bloom.Add(r.Context(), input.IdempotencyKey) } - // Emit audit event to stream (async, non-blocking path) - go func() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - event := domain.AuditEvent{ - EventID: uuid.New().String(), - ItemID: created.ID, - Action: "created", - Timestamp: time.Now().UTC(), - } - eventJSON, _ := json.Marshal(event) - _, _ = stream.Append(ctx, "audit-events", map[string]interface{}{ - "event": string(eventJSON), - }) - }() + auditEmitter.EmitAsync(domain.AuditEvent{ + EventID: uuid.New().String(), + ItemID: created.ID, + Action: "created", + Timestamp: time.Now().UTC(), + }) writeJSON(w, http.StatusCreated, envelope{"item": created}) } diff --git a/internal/handlers/items_create_test.go b/internal/handlers/items_create_test.go index f168b2e..dadfd44 100644 --- a/internal/handlers/items_create_test.go +++ b/internal/handlers/items_create_test.go @@ -4,10 +4,13 @@ import ( "bytes" "context" "encoding/json" + "io" + "log/slog" "net/http" "net/http/httptest" "testing" + "github.com/SShogun/redisforge/internal/audit" "github.com/SShogun/redisforge/internal/handlers" "github.com/SShogun/redisforge/internal/redisx" "github.com/SShogun/redisforge/internal/repo" @@ -24,9 +27,10 @@ func TestHandleCreateItem_Idempotency(t *testing.T) { } stream := redisx.NewStreamClient(client) + auditEmitter := audit.NewEmitter(stream, slog.New(slog.NewTextHandler(io.Discard, nil))) inMemoryRepo := repo.NewMemoryItemRepo() - handler := handlers.HandleCreateItem(inMemoryRepo, stream, bloom) + handler := handlers.HandleCreateItem(inMemoryRepo, auditEmitter, bloom) payload := map[string]interface{}{ "name": "Idempotency Test Widget", diff --git a/internal/handlers/items_delete.go b/internal/handlers/items_delete.go index 2f0e537..c084867 100644 --- a/internal/handlers/items_delete.go +++ b/internal/handlers/items_delete.go @@ -3,13 +3,16 @@ package handlers import ( "net/http" "strings" + "time" + "github.com/SShogun/redisforge/internal/audit" "github.com/SShogun/redisforge/internal/domain" "github.com/SShogun/redisforge/internal/repo" "github.com/go-chi/chi/v5" + "github.com/google/uuid" ) -func HandleDeleteItem(items repo.ItemRepo) http.HandlerFunc { +func HandleDeleteItem(items repo.ItemRepo, auditEmitter *audit.Emitter) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id := strings.TrimSpace(chi.URLParam(r, "id")) if id == "" { @@ -22,6 +25,13 @@ func HandleDeleteItem(items repo.ItemRepo) http.HandlerFunc { return } + auditEmitter.EmitAsync(domain.AuditEvent{ + EventID: uuid.New().String(), + ItemID: id, + Action: "deleted", + Timestamp: time.Now().UTC(), + }) + w.WriteHeader(http.StatusNoContent) } } diff --git a/internal/handlers/items_update.go b/internal/handlers/items_update.go index 7ff9984..d7d7c73 100644 --- a/internal/handlers/items_update.go +++ b/internal/handlers/items_update.go @@ -1,20 +1,18 @@ package handlers import ( - "context" - "encoding/json" "net/http" "strings" "time" + "github.com/SShogun/redisforge/internal/audit" "github.com/SShogun/redisforge/internal/domain" - "github.com/SShogun/redisforge/internal/redisx" "github.com/SShogun/redisforge/internal/repo" "github.com/go-chi/chi/v5" "github.com/google/uuid" ) -func HandleUpdateItem(items repo.ItemRepo, stream *redisx.StreamClient) http.HandlerFunc { +func HandleUpdateItem(items repo.ItemRepo, auditEmitter *audit.Emitter) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id := strings.TrimSpace(chi.URLParam(r, "id")) if id == "" { @@ -79,21 +77,12 @@ func HandleUpdateItem(items repo.ItemRepo, stream *redisx.StreamClient) http.Han return } - go func(item domain.Item) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - event := domain.AuditEvent{ - EventID: uuid.New().String(), - ItemID: item.ID, - Action: "updated", - Timestamp: time.Now().UTC(), - } - eventJSON, _ := json.Marshal(event) - _, _ = stream.Append(ctx, "audit-events", map[string]interface{}{ - "event": string(eventJSON), - }) - }(updatedItem) + auditEmitter.EmitAsync(domain.AuditEvent{ + EventID: uuid.New().String(), + ItemID: updatedItem.ID, + Action: "updated", + Timestamp: time.Now().UTC(), + }) writeJSON(w, http.StatusOK, envelope{"item": updatedItem}) } diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index 4dcb49d..4d320ba 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -47,6 +47,25 @@ var ( }, ) + // AuditEventsEmitted counts producer-side audit append attempts by action and result. + AuditEventsEmitted = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "audit_events_emitted_total", + Help: "Total producer-side audit event append attempts by action and status", + }, + []string{"action", "status"}, + ) + + // AuditEmitLatency measures producer-side audit serialization + XADD latency. + AuditEmitLatency = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "audit_emit_latency_ms", + Help: "Latency of producer-side audit event emission in milliseconds", + Buckets: []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}, + }, + []string{"action", "status"}, + ) + // StreamProcessingLatency measures end-to-end audit event processing time. StreamProcessingLatency = promauto.NewHistogramVec( prometheus.HistogramOpts{ @@ -125,6 +144,17 @@ func RecordCacheMiss() { CacheMisses.Inc() } +// RecordAuditEmit records producer-side audit event emission latency and result. +func RecordAuditEmit(start time.Time, action string, err error) { + elapsed := time.Since(start).Milliseconds() + status := "success" + if err != nil { + status = "error" + } + AuditEventsEmitted.WithLabelValues(action, status).Inc() + AuditEmitLatency.WithLabelValues(action, status).Observe(float64(elapsed)) +} + // RecordStreamProcessing records event processing latency and status. func RecordStreamProcessing(start time.Time, err error, action string) { elapsed := time.Since(start).Milliseconds()