Hardening: expose audit producer failures and document semantics

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,15 +51,17 @@ 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
 
     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
 

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
 

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.

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 ────────────────────────────────────────────────