open-lambda · tylerharter · Apr 23, 2026 · Apr 9, 2026 · Apr 9, 2026 · tylerharter
diff --git a/examples/kafka-db-sum/f.py b/examples/kafka-db-sum/f.py
@@ -0,0 +1,198 @@
+from flask import Flask, request, make_response, jsonify
+import os
+import random
+import psycopg2
+
+app = Flask(__name__)
+
+DATABASE_URL = os.environ.get("DATABASE_URL")
+
+# Probability (0.0–1.0) that a transaction will fail between UPDATE and COMMIT.
+# Set to 0 for normal operation; raise to stress-test seek-back recovery.
+FAIL_PROBABILITY = float(os.environ.get("FAIL_PROBABILITY", "0"))
+_db_initialized = False
+
+
+def get_db():
+    return psycopg2.connect(DATABASE_URL)
+
+
+def ensure_db():
+    """Create the running_sum table if it doesn't exist (runs once per sandbox)."""
+    global _db_initialized
+    if _db_initialized:
+        return
+    conn = get_db()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                CREATE TABLE IF NOT EXISTS running_sum (
+                    id INTEGER PRIMARY KEY DEFAULT 1,
+                    total BIGINT NOT NULL DEFAULT 0,
+                    last_offset BIGINT NOT NULL DEFAULT -1,
+                    message_count BIGINT NOT NULL DEFAULT 0,
+                    CHECK (id = 1)
+                )
+                """
+            )
+            cur.execute(
+                """
+                INSERT INTO running_sum (id, total, last_offset, message_count)
+                VALUES (1, 0, -1, 0)
+                ON CONFLICT (id) DO NOTHING
+                """
+            )
+        conn.commit()
+        _db_initialized = True
+    finally:
+        conn.close()
+
+
+@app.route("/reset", methods=["POST"])
+def reset():
+    """Reset running_sum to zero so the demo can be re-run cleanly."""
+    ensure_db()
+    conn = get_db()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                "UPDATE running_sum SET total = 0, last_offset = -1, "
+                "message_count = 0 WHERE id = 1"
+            )
+        conn.commit()
+        return jsonify({"status": "reset"})
+    finally:
+        conn.close()
+
+
+@app.route("/", methods=["GET", "POST"])
+def handle():
+    ensure_db()
+
+    # GET — return current state (useful for checking progress via HTTP)
+    if request.method == "GET":
+        conn = get_db()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT total, last_offset, message_count "
+                    "FROM running_sum WHERE id = 1"
+                )
+                row = cur.fetchone()
+                if row:
+                    return jsonify(
+                        {
+                            "running_sum": row[0],
+                            "last_offset": row[1],
+                            "message_count": row[2],
+                        }
+                    )
+                return jsonify(
+                    {"running_sum": 0, "last_offset": -1, "message_count": 0}
+                )
+        finally:
+            conn.close()
+
+    # POST — process a Kafka message containing a number
+    offset = int(request.headers.get("X-Kafka-Offset", "-1"))
+    topic = request.headers.get("X-Kafka-Topic", "unknown")
+    partition = request.headers.get("X-Kafka-Partition", "unknown")
+
+    body = request.get_json(silent=True)
+
+    # Accept {"number": N}
+    if isinstance(body, dict):
+        number = body.get("number", 0)
+
+    conn = None
+    try:
+        conn = get_db()
+        with conn.cursor() as cur:
+            # Lock the row and read last processed offset
+            cur.execute(
+                "SELECT last_offset FROM running_sum WHERE id = 1 FOR UPDATE"
+            )
+            row = cur.fetchone()
+            last_offset = row[0] if row else -1
+
+            # Idempotency: skip if this offset was already processed.
+            # This prevents double-counting after a seek-back replays
+            # messages that were already committed.
+            if offset <= last_offset:
+                conn.rollback()
+                print(f"[skip] offset={offset} already processed (last={last_offset})")
+                return jsonify(
+                    {
+                        "status": "skipped",
+                        "reason": "already processed",
+                        "offset": offset,
+                        "last_offset": last_offset,
+                    }
+                )
+
+            # Atomically add number to running sum and advance the offset
+            cur.execute(
+                """
+                UPDATE running_sum
+                SET total = total + %s,
+                    last_offset = %s,
+                    message_count = message_count + 1
+                WHERE id = 1
+                """,
+                (number, offset),
+            )
+
+            # --- Fault injection ------------------------------------------------
+            # Simulate a crash between UPDATE and COMMIT.  
+            if FAIL_PROBABILITY > 0 and random.random() < FAIL_PROBABILITY:
+                raise Exception(
+                    f"Simulated DB failure at offset {offset} "
+                    f"(FAIL_PROBABILITY={FAIL_PROBABILITY})"
+                )
+            # --------------------------------------------------------------------
+
+            conn.commit()
+
+            # Read back the new state for the response
+            cur.execute(
+                "SELECT total, last_offset, message_count "
+                "FROM running_sum WHERE id = 1"
+            )
+            total, last_off, count = cur.fetchone()
+
+        print(f"[ok] offset={offset} number={number} sum={total} count={count}")
+        return jsonify(
+            {
+                "status": "ok",
+                "offset": offset,
+                "number_added": number,
+                "running_sum": total,
+                "message_count": count,
+            }
+        )
+
+    except Exception as e:
+        print(f"[error] offset={offset} error={e}")
+        if conn:
+            try:
+                conn.rollback()
+            except Exception:
+                pass
+
+        # Tell OL's Kafka consumer to seek back to this offset and retry.
+        # The consumer's LRU cache will serve the replay without re-fetching
+        # from Kafka, and the idempotency check above prevents double-counting
+        # for any offsets that were already committed before the failure.
+        resp = make_response(
+            jsonify({"status": "error", "offset": offset, "error": str(e)}), 500
+        )
+        resp.headers["X-Kafka-Seek-Offset"] = str(offset)
+        return resp
+
+    finally:
+        if conn:
+            try:
+                conn.close()
+            except Exception:
+                pass
diff --git a/examples/kafka-db-sum/instructions.md b/examples/kafka-db-sum/instructions.md
@@ -0,0 +1,106 @@
+# kafka-db-sum: Testing Instructions
+
+## Prerequisites
+
+- OpenLambda built (`make ol imgs/ol-min`)
+- Docker installed
+
+## 1. Start PostgreSQL
+
+```bash
+docker run -d --name ol-pg \
+  --network host \
+  -e POSTGRES_USER=ol \
+  -e POSTGRES_PASSWORD=ol \
+  -e POSTGRES_DB=ol_demo \
+  postgres:16
+```
+
+## 2. Start Kafka
+
+```bash
+docker run -d --name kafka \
+  -p 9092:9092 \
+  apache/kafka:latest
+```
+
+## 3. Create the `numbers` topic
+
+```bash
+docker exec kafka /opt/kafka/bin/kafka-topics.sh --create \
+  --topic numbers \
+  --bootstrap-server localhost:9092
+```
+
+## 4. Initialize and start the OL worker
+
+From the repository root:
+
+```bash
+sudo -A ./ol worker init -p ../default-ol -i ol-min
+sudo -A ./ol worker up -p ../default-ol
+```
+
+Run `worker up` in a separate terminal, or add `-d` for detached mode.
+The worker listens on `localhost:5000` by default.
+
+## 5. Install the lambda
+
+From the repository root:
+
+```bash
+./ol admin install examples/kafka-db-sum/
+```
+
+## 6. Register the Kafka consumer
+
+A standalone worker does not auto-register Kafka triggers on upload.
+Register manually:
+
+```bash
+curl -X POST localhost:5000/kafka/register/kafka-db-sum
+```
+
+## 7. Send test messages
+
+Python producer script (requires `pip install kafka-python`):
+
+```bash
+python examples/kafka-db-sum/produce.py 100
+```
+
+## 8. Check results
+
+```bash
+curl localhost:5000/run/kafka-db-sum/
+```
+
+Expected output (sum of 1..100 = 5050):
+
+```json
+{ "last_offset": 99, "message_count": 100, "running_sum": 5050 }
+```
+
+## 9. Reset and re-run
+
+```bash
+curl -X POST localhost:5000/run/kafka-db-sum/reset
+```
+
+Then send a fresh batch (step 7) and verify again.
+
+## Configuration
+
+In `ol.yaml`:
+
+| Variable           | Default                                     | Description                                                         |
+| ------------------ | ------------------------------------------- | ------------------------------------------------------------------- |
+| `DATABASE_URL`     | `postgresql://ol:ol@127.0.0.1:5432/ol_demo` | PostgreSQL connection string                                        |
+| `FAIL_PROBABILITY` | `0`                                         | Chance (0.0-1.0) of simulated failure. Use `0.3` to test seek-back. |
+
+## Cleanup
+
+```bash
+sudo -A ./ol worker down -p default-ol
+docker rm -f kafka ol-pg
+```
diff --git a/examples/kafka-db-sum/ol.yaml b/examples/kafka-db-sum/ol.yaml
@@ -0,0 +1,16 @@
+triggers:
+  http:
+    - method: GET
+    - method: POST
+  kafka:
+    - bootstrap_servers:
+        - "localhost:9092"
+      topics:
+        - "numbers"
+      auto_offset_reset: "earliest"
+
+environment:
+  DATABASE_URL: "postgresql://ol:ol@127.0.0.1:5432/ol_demo"
+  # Probability (0.0-1.0) of simulated DB failure between UPDATE and COMMIT.
+  # Set to "0" for normal operation. Try "0.3" to see seek-back recovery in action.
+  FAIL_PROBABILITY: "0.3"
diff --git a/examples/kafka-db-sum/produce.py b/examples/kafka-db-sum/produce.py
@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""
+Send numbered messages to the 'numbers' Kafka topic.
+
+Usage:
+    python produce.py              # send numbers 1..10
+    python produce.py 100          # send numbers 1..100
+    python produce.py 50 0.5       # send 1..50 with 0.5s delay between each
+"""
+
+import json
+import sys
+import time
+
+from kafka import KafkaProducer
+
+BROKER = "localhost:9092"
+TOPIC = "numbers"
+
+
+def main():
+    count = int(sys.argv[1]) if len(sys.argv) > 1 else 10
+    delay = float(sys.argv[2]) if len(sys.argv) > 2 else 0.1
+
+    producer = KafkaProducer(
+        bootstrap_servers=BROKER,
+        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
+    )
+
+    expected_sum = 0
+    print(f"Sending numbers 1..{count} to topic '{TOPIC}'...")
+    for i in range(1, count + 1):
+        producer.send(TOPIC, {"number": i})
+        expected_sum += i
+        print(f"  sent {i}")
+        if delay:
+            time.sleep(delay)
+
+    producer.flush()
+    producer.close()
+    print(f"\nDone. Expected sum = {expected_sum}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/kafka-db-sum/requirements.in b/examples/kafka-db-sum/requirements.in
@@ -0,0 +1,3 @@
+flask==2.3.2
+werkzeug==3.0.3
+psycopg2-binary==2.9.9
diff --git a/examples/kafka-db-sum/requirements.txt b/examples/kafka-db-sum/requirements.txt
@@ -0,0 +1,26 @@
+#
+# This file is autogenerated by pip-compile with Python 3.13
+# by the following command:
+#
+#    pip-compile requirements.in
+#
+blinker==1.6.2
+    # via flask
+click==8.1.7
+    # via flask
+flask==2.3.2
+    # via -r requirements.in
+itsdangerous==2.1.2
+    # via flask
+jinja2==3.1.4
+    # via flask
+markupsafe==2.1.3
+    # via
+    #   jinja2
+    #   werkzeug
+psycopg2-binary==2.9.9
+    # via -r requirements.in
+werkzeug==3.0.3
+    # via
+    #   -r requirements.in
+    #   flask