Streaming mapper, CQRS read-replica, shuffle integrity, benchmarks, and bug fixes

.gitignore

@@ -188,3 +188,12 @@ benchmarks/data/.swarm/
 __pycache__/
 *.py[cod]
 *$py.class
+
+# Temporary local files
+scratchpad.md
+demo_3m_guide.md
+.swarm/
+docs/superpowers/
+testdata/big-file/input.jsonl
+testdata/job-fail/input.jsonl
+

README.md

@@ -150,6 +150,7 @@ All protected routes require `Authorization: Bearer <token>`.
 | `POST` | `/api/v1/admin/users` | `ADMIN` |
 | `DELETE` | `/api/v1/admin/users/{username}` | `ADMIN` |
 | `POST` | `/api/v1/admin/config/workers` | `ADMIN` |
+| `GET`  | `/api/v1/admin/config/workers` | `ADMIN` |
 
 ## gRPC contract (manager <-> worker)
 
@@ -166,14 +167,59 @@ Regenerate stubs after editing proto:
 protoc --go_out=. --go-grpc_out=. proto/mapreduce.proto
 ```
 
-## Documentation
+## Benchmarking & Verification
+
+KubeMapReduce includes a comprehensive benchmarking suite and failure-injection tests to validate the system's scalability and resilience.
+
+### Performance Benchmarks
+
+Apache Bench results and distributed scalability data are available in `benchmarks/results/`:
+
+| Metric | Value |
+|--------|-------|
+| Throughput (Apache Bench, C=20, n=500) | 20.43 req/s |
+| Combiner improvement (no combiner vs with combiner) | 69s → 47s (33% reduction) |
+| Grep scalability (1→8 reducers) | 20s → 44s (sub-linear due to shuffle overhead) |
+| Worker kill recovery | 20s total (15s detection + 5s reschedule) |
+| Manager restart recovery | 22s total (5s pod term + 15s startup + 2s state recovery) |
+| Zombie fencing (lease-based) | 30s lease expiry, data integrity preserved |
+
+**Run distributed benchmarks on a live cluster:**
+
+```bash
+# Prerequisites: logged in via CLI, API_URL set
+export API_URL="http://<YOUR_API_IP>"
+python benchmarks/distributed_benchmark.py
+```
+
+**Run local validation (WordCount with combiner):**
+
+```bash
+python benchmarks/wordcount/validate.py
+```
+
+**Run local validation (Grep):**
+
+```bash
+python benchmarks/grep/validate.py --pattern "love"
+```
+
+### Cluster Deployment (GKE)
+
+For a complete production-grade GKE deployment with managed PostgreSQL, MinIO, and Keycloak:
+
+- **[docs/DEPLOYMENT.md](docs/DEPLOYMENT.md)** — Step-by-step GKE deployment guide
+- **[docs/CLUSTER_DEPLOYMENT.md](docs/CLUSTER_DEPLOYMENT.md)** — 10-step production checklist
+- **[docs/PRESENTATION_TESTS.md](docs/PRESENTATION_TESTS.md)** — E2E failure injection results
+- **[docs/EXTENDED_BENCHMARKS.md](docs/EXTENDED_BENCHMARKS.md)** — Scalability and concurrency proofs
+- **[docs/MONITORING_AND_RESULTS.md](docs/MONITORING_AND_RESULTS.md)** — Prometheus/Grafana dashboards
+
+### Docs Quick Reference
 
 - **Architecture**: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md)
 - **Data locality architecture**: [docs/ARCHITECTURE_DATA_LOCALITY.md](docs/ARCHITECTURE_DATA_LOCALITY.md)
 - **External access architecture**: [docs/ARCHITECTURE_EXTERNAL_ACCESS.md](docs/ARCHITECTURE_EXTERNAL_ACCESS.md)
-- **Cluster deployment checklist**: [docs/CLUSTER_DEPLOYMENT.md](docs/CLUSTER_DEPLOYMENT.md)
 - **Minikube local flow**: [docs/MINIKUBE_LOCAL_DEV.md](docs/MINIKUBE_LOCAL_DEV.md)
 - **Linkerd setup**: [docs/LINKERD_SETUP.md](docs/LINKERD_SETUP.md)
 - **Timeout tuning**: [docs/TIMEOUT_CONFIGURATION.md](docs/TIMEOUT_CONFIGURATION.md)
-- **Monitoring/results**: [docs/MONITORING_AND_RESULTS.md](docs/MONITORING_AND_RESULTS.md)
 - **Migrations guide**: [migrations/README.md](migrations/README.md)

auth-service/cmd/setup/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"log"
@@ -44,7 +45,7 @@ func main() {
 	// Optional: create initial user
 	username := flag.String("username", "", "Username to create after bootstrap (optional)")
 	email := flag.String("email", "", "Email for the new user")
-	password := flag.String("password", "", "Password for the new user")
+	password := flag.String("password", getEnv("SETUP_USER_PASSWORD", ""), "Password for the new user (or SETUP_USER_PASSWORD env)")
 	promptPassword := flag.Bool("prompt-password", false, "Prompt securely for new user password (input hidden)")
 	role := flag.String("role", "ADMIN", "Role to assign: ADMIN or USER")
 
@@ -110,7 +111,11 @@ func main() {
 		Password: userPassword,
 		Role:     normalizedRole,
 	}); err != nil {
-		log.Fatalf("failed to create user: %v", err)
+		if errors.Is(err, auth.ErrUserAlreadyExists) {
+			fmt.Printf("==> User %q already exists, skipping user creation.\n", strings.TrimSpace(*username))
+		} else {
+			log.Fatalf("failed to create user: %v", err)
+		}
 	}
 
 	fmt.Printf("==> Created %s user %q in realm %q.\n", normalizedRole, strings.TrimSpace(*username), cfg.Realm)

auth-service/pkg/auth/keycloak_admin.go

@@ -14,6 +14,11 @@ import (
 	"time"
 )
 
+// ErrUserAlreadyExists is returned by CreateUser when the Keycloak user
+// already exists (HTTP 409). Callers should check with errors.Is to avoid
+// brittle string matching.
+var ErrUserAlreadyExists = errors.New("user already exists")
+
 const (
 	defaultAdminHTTPTimeout = 10 * time.Second
 	adminTokenRefreshSkew   = 30 * time.Second
@@ -129,6 +134,9 @@ func (c *KeycloakAdminClient) CreateUser(ctx context.Context, req CreateUserRequ
 	}
 	defer httpResp.Body.Close()
 
+	if httpResp.StatusCode == http.StatusConflict {
+		return ErrUserAlreadyExists
+	}
 	if err := ensureStatus(httpResp, http.StatusCreated, "create user"); err != nil {
 		return err
 	}

auth-service/pkg/auth/token_store.go

@@ -15,10 +15,11 @@ import (
 // invocations. The refresh token is used to obtain new access tokens when the
 // current one expires, reducing the frequency of interactive logins.
 type StoredTokens struct {
-	AccessToken  string `json:"access_token"`
-	RefreshToken string `json:"refresh_token"`
-	ExpiresAt    int64  `json:"expires_at"`
-	ServerURL    string `json:"server_url"`
+	AccessToken     string `json:"access_token"`
+	RefreshToken    string `json:"refresh_token"`
+	ExpiresAt       int64  `json:"expires_at"`
+	ServerURL       string `json:"server_url"`
+	KeycloakBaseURL string `json:"keycloak_base_url,omitempty"`
 }
 
 // IsAccessExpired checks if the access token has expired.

auth-service/pkg/auth/token_store_test.go

@@ -56,10 +56,11 @@ func TestSaveAndLoadTokens(t *testing.T) {
 	}()
 
 	original := &StoredTokens{
-		AccessToken:  "acc-tok",
-		RefreshToken: "ref-tok",
-		ExpiresAt:    time.Now().Unix() + 300,
-		ServerURL:    "http://localhost:8081",
+		AccessToken:     "acc-tok",
+		RefreshToken:    "ref-tok",
+		ExpiresAt:       time.Now().Unix() + 300,
+		ServerURL:       "http://localhost:8081",
+		KeycloakBaseURL: "http://keycloak:8080",
 	}
 
 	if err := SaveTokens(original); err != nil {
@@ -174,10 +175,11 @@ func TestLoadTokens_InvalidJSON(t *testing.T) {
 
 func TestStoredTokens_JSONRoundTrip(t *testing.T) {
 	original := StoredTokens{
-		AccessToken:  "at",
-		RefreshToken: "rt",
-		ExpiresAt:    1234567890,
-		ServerURL:    "http://example.com",
+		AccessToken:     "at",
+		RefreshToken:    "rt",
+		ExpiresAt:       1234567890,
+		ServerURL:       "http://example.com",
+		KeycloakBaseURL: "http://keycloak:8080",
 	}
 
 	data, err := json.Marshal(original)

benchmarks/README.md

@@ -1,26 +1,28 @@
-# Benchmarks
+# KubeMapReduce Benchmarks
 
-Two benchmark suites validating the MapReduce pipeline: **WordCount** and **Distributed Grep**.
+This directory contains the benchmarks used to validate the performance and resilience of the KubeMapReduce system.
 
-## Prerequisites
+## Dataset Preparation
 
-- Python 3 (no third-party packages required)
-- 6 Project Gutenberg HTML files in `corpus/` (already present in this repo)
-
-## Step 1 — Build the corpus
-
-Strips HTML from the Gutenberg books and produces a JSONL dataset:
+To generate the datasets used in these benchmarks, run the following commands:
 
+### 1. Build the base corpus (16MB)
+Requires the Gutenberg HTML files in `corpus/`.
 ```bash
 python benchmarks/build_corpus.py
 ```
-
 Output: `benchmarks/data/corpus.jsonl` (~77,000 lines, one per text line from each book).
 
 Each record: `{"key": "pg1342:000042", "value": "It is a truth universally acknowledged"}`
 
 The `benchmarks/data/` directory is gitignored. Re-run this command after cloning.
 
+### 2. Build the Big Data corpus (128MB)
+Requires `benchmarks/data/corpus.jsonl` to exist.
+```bash
+python -c "with open('benchmarks/data/corpus.jsonl', 'r') as f: content = f.read(); open('benchmarks/data/corpus_128mb.jsonl', 'w').write(content * 8)"
+```
+
 ---
 
 ## WordCount
@@ -41,22 +43,6 @@ Runs the full pipeline locally (no cluster needed) and checks correctness:
 python benchmarks/wordcount/validate.py
 ```
 
-Expected output:
-```
-Running WordCount pipeline locally against .../corpus.jsonl ...
-  12345 unique words found
-  Top 10 words: the=18432, and=14201, of=12800, ...
-PASS
-```
-
-### Compare against cluster output
-
-After running the job on a live cluster and downloading the output:
-
-```bash
-python benchmarks/wordcount/validate.py --compare /path/to/cluster-output.jsonl
-```
-
 ---
 
 ## Distributed Grep
@@ -96,7 +82,21 @@ python benchmarks/grep/validate.py --compare /path/to/cluster-output.jsonl --pat
 
 ---
 
-## Running on a live cluster
+## Running Benchmarks on a Live Cluster (GKE)
+
+### Prerequisites
+Ensure you have logged in via the CLI and set the `API_URL` environment variable.
+```bash
+export API_URL="http://<YOUR_API_IP>"
+```
+
+### Distributed GKE Benchmark Suite
+To run the automated performance and scalability benchmarks:
+```bash
+python benchmarks/distributed_benchmark.py
+```
+
+### Manual Job Submission
 
 1. **Build corpus** (Step 1 above)
 
@@ -119,6 +119,12 @@ python benchmarks/grep/validate.py --compare /path/to/cluster-output.jsonl --pat
 
 ---
 
+## Results & Visualizations
+Final results and visualizations are located in `benchmarks/results/`.
+Detailed analysis can be found in `docs/PRESENTATION_TESTS.md` and `docs/EXTENDED_BENCHMARKS.md`.
+
+---
+
 ## Mapper/Combiner/Reducer Protocol
 
 Mappers, combiners, and reducers communicate via JSONL on stdin/stdout: