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()