Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ CleanStack cleans tabular files and documents with AI suggestions and human appr
- **Pipelines and templates:** save the approved rules of a pipeline as a template and reuse it. A chat builder can draft pipelines and synthetic data.
- **Schema drift alerts:** optional Slack webhook per pipeline.
- **Usage controls:** monthly row quotas per plan (Free: 50,000 rows; plans are assigned by an admin, with no payment integration), Upstash rate limits, and an AI spend cap checked in Postgres before every Bedrock call.
- **Guest access:** **Try as guest** on the landing and sign-in pages starts a 24-hour session with no account. **Try with sample data** loads a seeded demo template, so it makes no Bedrock call. Guests get 2 MB files, 3 uploads, 5,000 rows per run and 10 AI calls per hour, with $0.25 of AI spend per guest. AI-heavy or irreversible features (chat builder, auto-clean, training export, Slack alerts, templates, account deletion) are blocked for guests. A guest's data is erased after 24 hours. [Full limits](docs/security-model.md#guest-access)
- **Guest access:** **Try as guest** on the landing and sign-in pages starts a 24-hour session with no account. **Try with sample data** loads a seeded demo template, so it makes no Bedrock call. Guests get 2 MB files, 3 uploads, 5,000 rows per run and 10 AI calls per hour, with $0.25 of AI spend per guest. AI-heavy or irreversible features (chat builder, auto-clean, training export, Slack alerts, templates, account deletion) are blocked for guests. A guest's data is erased by the purge that runs every 4 hours, once the 24-hour session has ended. [Full limits](docs/security-model.md#guest-access)

## Architecture

Expand All @@ -41,7 +41,7 @@ Data PR approval / auto-clean ──► SQS executor queue ──► Lambda exec

Other services: Clerk (auth), Upstash Redis (rate limits and quota cache, which fail open when not configured) and Sentry (optional). AWS resources are configured by hand, and there is no infrastructure as code in this repo.

A run moves through `pending → profiling → awaiting_ai → awaiting_approval → queued → running → completed | failed`. Execution is idempotent per run: the executor claims a run with a conditional `UPDATE`, so a duplicate SQS delivery is skipped. A stale run is marked failed after 20 minutes by `/api/cron/reconcile-runs`.
A run moves through `pending → profiling → awaiting_ai → awaiting_approval → queued → running → completed | failed`. Execution is idempotent per run: the executor claims a run with a conditional `UPDATE`, so a duplicate SQS delivery is skipped. A run stuck for 20 minutes is marked failed by `/api/cron/reconcile-runs`, which runs every 4 hours.

## Local development

Expand Down Expand Up @@ -96,7 +96,7 @@ Vercel builds and deploys `main`. The Lambdas, migrations and AWS resources are
- applying migrations 001–003 (`run-migration.mjs`, or the RDS Data API from CloudShell);
- the CloudShell zip-swap deploy for `cleanstack-profiler`, `cleanstack-executor`, `cleanstack-ai-trigger` and `cleanstack-drift`;
- raw bucket CORS (PUT, GET, POST), the profiler trigger on all object create events, and processed bucket CORS;
- EventBridge rules `cleanstack-reconciler-5min` (every 5 minutes) and `cleanstack-purge-guests-hourly` (every hour), which call the cron routes through API destinations as the `cleanstack-eventbridge-invoker` role;
- EventBridge rules `cleanstack-reconciler-5min` and `cleanstack-purge-guests-hourly`, both now `cron(0 */4 * * ? *)` (every 4 hours, so Aurora can auto-pause), which call the cron routes through API destinations as the `cleanstack-eventbridge-invoker` role;
- the S3 lifecycle rule on the `guest_` prefix;
- least-privilege IAM for the `cleanstack-vercel` user and for the Lambda roles.

Expand Down
6 changes: 4 additions & 2 deletions docs/deployment.md
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,10 @@ The profiler, ai-trigger and web app must share the same `WEBHOOK_SECRET`.

| Rule | Schedule | API destination | Route |
|---|---|---|---|
| `cleanstack-reconciler-5min` | `rate(5 minutes)` | `cleanstack-reconciler-destination` | `/api/cron/reconcile-runs`: marks runs stuck for 20 minutes as `failed` |
| `cleanstack-purge-guests-hourly` | `rate(1 hour)` | `cleanstack-purge-guests-destination` | `/api/cron/purge-guests`: erases up to 20 expired guests per call |
| `cleanstack-reconciler-5min` | `cron(0 */4 * * ? *)` | `cleanstack-reconciler-destination` | `/api/cron/reconcile-runs`: marks runs stuck for 20 minutes as `failed` |
| `cleanstack-purge-guests-hourly` | `cron(0 */4 * * ? *)` | `cleanstack-purge-guests-destination` | `/api/cron/purge-guests`: erases up to 20 expired guests per call |

Both rules run every 4 hours (00:00, 04:00, ... UTC). The names are from the original 5-minute and hourly schedules. The reconciler used to wake the database every 5 minutes, so the Aurora Serverless v2 cluster `database-1` (minimum 0 ACU, auto-pause after 300 s) never paused, which cost about $45 a month. Now the cluster is paused most of the time, and the first request after a pause waits while it resumes (see [Aurora auto-pause](operations.md#aurora-auto-pause)).

If you rotate `CRON_SECRET`, update the connection `cleanstack-reconciler-auth` at the same time as Vercel.

Expand Down
15 changes: 13 additions & 2 deletions docs/operations.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

| Symptom | What to check or do |
|---|---|
| Run stuck in `profiling`, `awaiting_ai`, `queued` or `running` | `reconcile-runs` marks these runs `failed` after 20 minutes, so first confirm that the EventBridge rule `cleanstack-reconciler-5min` is firing. |
| Run stuck in `profiling`, `awaiting_ai`, `queued` or `running` | `reconcile-runs` marks runs stuck for 20 minutes as `failed`. It runs every 4 hours (`cron(0 */4 * * ? *)`), so a stuck run can stay that way for up to about 4 h 20 min. First confirm that the EventBridge rule `cleanstack-reconciler-5min` is firing. |
| Run `failed` with "Could not queue this run …" | Sending to SQS failed (log `SQS send failed; marking run failed` or `AI jobs SQS send failed; marking run failed`), so the route failed the run and returned 503. Check the queue URL and the `sqs:SendMessage` permission of `cleanstack-vercel`. The user retries by uploading again. |
| Runs stay in `pending` after upload | The profiler trigger is missing `Post`. It must fire on all object create events. |
| Run `failed` with `Attempt n/3 failed, retrying` in its history | SQS retries transient errors up to `EXECUTOR_MAX_ATTEMPTS` times, and the last attempt marks the run failed. Look up the run id in the executor's CloudWatch logs. |
Expand All @@ -16,4 +16,15 @@
| Rotating `WEBHOOK_SECRET` | Update Vercel and the profiler and ai-trigger Lambdas together. Requests fail with 401 while the values differ. |
| AI spend | `GET /api/admin/ai-spend` with `x-admin-secret` lists this month's estimated spend per team. |
| Guests see "Guest AI capacity is used up for today" | All guests together have reached $5 of estimated Bedrock spend since 00:00 UTC. The cap resets at midnight UTC (`GUEST_LIMITS.aiSpendAllGuestsPerDayUsd`). |
| Guest data not being deleted | Check the rule `cleanstack-purge-guests-hourly` and its `invoke-purge-guests` policy. Each call erases up to 20 expired guests. The `guest_` lifecycle rule removes S3 objects after 1 day even if the purge stops, but the database rows remain until the purge runs. |
| Guest data not being deleted | Check the rule `cleanstack-purge-guests-hourly` (now every 4 hours, `cron(0 */4 * * ? *)`) and its `invoke-purge-guests` policy. Each call erases up to 20 expired guests, so about 120 a day. The `guest_` lifecycle rule removes S3 objects after 1 day even if the purge stops, but the database rows remain until the purge runs. |

## Aurora auto-pause

The cluster `database-1` (Aurora Serverless v2, PostgreSQL 17.7) has a minimum of 0 ACU and pauses after 300 s without connections. It used to never pause, because the reconciler called the database every 5 minutes, and that cost about $45 a month. Both EventBridge rules (`cleanstack-reconciler-5min` and `cleanstack-purge-guests-hourly`) now run on `cron(0 */4 * * ? *)`, every 4 hours, so the cluster is paused most of the time.

The first request after a pause starts the resume, which takes about 15 s, or 30 s or more after a pause longer than 24 h. Callers wait instead of failing:

- **Web app (RDS Data API):** while the cluster resumes, the Data API answers `DatabaseResumingException`. `src/lib/db-resume.ts` retries only that error, for `ExecuteStatement`, `BeginTransaction`, `CommitTransaction` and `RollbackTransaction`, with backoff of 1, 2, 4 and then 8 s, for up to 35 s. AWS documents the request as cancelled before it ran, so retrying writes is safe. Any other error, including statement timeouts, is never retried. The log line is `database is resuming from auto-pause; retrying Data API call` (warn, with `op`). The cron routes have `maxDuration` 60 (reconcile) and 120 (purge) to leave room for the wait.
- **Lambdas (psycopg2, IAM auth):** profiler, executor and drift connect directly. A connection to a paused cluster normally waits until it has resumed. A failed connect (`OperationalError`, except authentication failures) is retried for up to 35 s. The CloudWatch line is `[db] connect failed, database may be resuming from auto-pause`. ai-trigger does not connect to the database.

If a request still fails with `DatabaseResumingException` after a long pause, the resume took longer than the 35 s budget. The constants are `RESUME_RETRY_BUDGET_MS` in `src/lib/db-resume.ts` and `DB_CONNECT_RETRY_BUDGET_S` in each Lambda.
2 changes: 1 addition & 1 deletion docs/security-model.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,6 @@ Guest access is off until `GUEST_COOKIE_SECRET` is set. `POST /api/guest` checks
| AI spend | $0.25 per guest, and $5 per UTC day for all guests together | `checkAiBudget` |
| Sessions | 5 per IP hash, and 200 overall, per 24 h | `POST /api/guest` |

The values live in `src/lib/guest-limits.ts`. Guests cannot use the chat builder or synthetic data, auto-clean, training export, Slack alerts, account deletion, templates or `/api/admin`. The middleware answers 403 for these, or redirects `/templates` to the dashboard, and each route also rejects guest ids (`forbidGuest`). **Try with sample data** (`POST /api/sample-data`) uploads the CSV from `src/lib/sample-data.ts` and attaches the demo template that migration 003 seeds, so it makes no Bedrock call. After 24 h the cookie stops working, and `/api/cron/purge-guests` erases the guest's data. Guest work is not carried over to a new account on sign-up.
The values live in `src/lib/guest-limits.ts`. Guests cannot use the chat builder or synthetic data, auto-clean, training export, Slack alerts, account deletion, templates or `/api/admin`. The middleware answers 403 for these, or redirects `/templates` to the dashboard, and each route also rejects guest ids (`forbidGuest`). **Try with sample data** (`POST /api/sample-data`) uploads the CSV from `src/lib/sample-data.ts` and attaches the demo template that migration 003 seeds, so it makes no Bedrock call. After 24 h the cookie stops working, and the next run of `/api/cron/purge-guests` (every 4 hours) erases the guest's data. Guest work is not carried over to a new account on sign-up.

Report vulnerabilities as described in [SECURITY.md](../SECURITY.md).
28 changes: 27 additions & 1 deletion lambdas/drift/handler.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,38 @@
import json
import os
import time
import boto3
import psycopg2
import requests

secrets = boto3.client("secretsmanager")


# Aurora Serverless v2 auto-pause: a connection to a paused cluster normally waits while it
# resumes (~15s, 30s+ after a pause longer than 24h), but AWS recommends retrying failed
# connections, since a burst of connections during a resume can fail. A failed connect ran no
# SQL, so retrying it is safe. Authentication errors are not retried. Same helper in every
# Lambda that connects (profiler, executor, drift); each Lambda ships as one handler.py.
DB_CONNECT_RETRY_BUDGET_S = 35


def _connect_with_resume_retry(**kwargs):
deadline = time.monotonic() + DB_CONNECT_RETRY_BUDGET_S
delay, logged = 1, False
while True:
try:
return psycopg2.connect(**kwargs)
except psycopg2.OperationalError as e:
if "authentication failed" in str(e) or time.monotonic() + delay > deadline:
raise
if not logged:
print(f"[db] connect failed, database may be resuming from auto-pause; retrying "
f"for up to {DB_CONNECT_RETRY_BUDGET_S}s: {str(e).strip()[:200]}")
logged = True
time.sleep(delay)
delay = min(delay * 2, 8)


def get_db_conn():
secret = json.loads(
secrets.get_secret_value(SecretId=os.environ["DB_SECRET_ARN"])["SecretString"]
Expand All @@ -17,7 +43,7 @@ def get_db_conn():
dbname = secret.get("dbname", "cleanstack")
rds = boto3.client("rds", region_name=os.environ.get("AWS_REGION", "us-east-1"))
token = rds.generate_db_auth_token(DBHostname=host, Port=port, DBUsername=user)
return psycopg2.connect(host=host, port=port, user=user, password=token, dbname=dbname, sslmode="require")
return _connect_with_resume_retry(host=host, port=port, user=user, password=token, dbname=dbname, sslmode="require")


def compute_diff(old: dict, new: dict) -> dict:
Expand Down
27 changes: 26 additions & 1 deletion lambdas/executor/handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -557,6 +557,31 @@ def _parse_params_simple(raw) -> dict:
return raw or {}


# Aurora Serverless v2 auto-pause: a connection to a paused cluster normally waits while it
# resumes (~15s, 30s+ after a pause longer than 24h), but AWS recommends retrying failed
# connections, since a burst of connections during a resume can fail. A failed connect ran no
# SQL, so retrying it is safe. Authentication errors are not retried. Same helper in every
# Lambda that connects (profiler, executor, drift); each Lambda ships as one handler.py.
DB_CONNECT_RETRY_BUDGET_S = 35


def _connect_with_resume_retry(**kwargs):
deadline = time.monotonic() + DB_CONNECT_RETRY_BUDGET_S
delay, logged = 1, False
while True:
try:
return psycopg2.connect(**kwargs)
except psycopg2.OperationalError as e:
if "authentication failed" in str(e) or time.monotonic() + delay > deadline:
raise
if not logged:
print(f"[db] connect failed, database may be resuming from auto-pause; retrying "
f"for up to {DB_CONNECT_RETRY_BUDGET_S}s: {str(e).strip()[:200]}")
logged = True
time.sleep(delay)
delay = min(delay * 2, 8)


def get_db_conn():
secret = json.loads(
secrets.get_secret_value(SecretId=os.environ["DB_SECRET_ARN"])["SecretString"]
Expand All @@ -567,7 +592,7 @@ def get_db_conn():
dbname = secret.get("dbname", "cleanstack")
rds = boto3.client("rds", region_name=os.environ.get("AWS_REGION", "us-east-1"))
token = rds.generate_db_auth_token(DBHostname=host, Port=port, DBUsername=user)
return psycopg2.connect(host=host, port=port, user=user, password=token, dbname=dbname, sslmode="require")
return _connect_with_resume_retry(host=host, port=port, user=user, password=token, dbname=dbname, sslmode="require")


def load_raw_dataframe(file_bytes: bytes, fmt: str) -> pd.DataFrame:
Expand Down
28 changes: 27 additions & 1 deletion lambdas/profiler/handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import json
import io
import re
import time
import boto3
import psycopg2
import requests
Expand Down Expand Up @@ -299,14 +300,39 @@ def compute_quality_score(df: pd.DataFrame) -> dict:
}


# Aurora Serverless v2 auto-pause: a connection to a paused cluster normally waits while it
# resumes (~15s, 30s+ after a pause longer than 24h), but AWS recommends retrying failed
# connections, since a burst of connections during a resume can fail. A failed connect ran no
# SQL, so retrying it is safe. Authentication errors are not retried. Same helper in every
# Lambda that connects (profiler, executor, drift); each Lambda ships as one handler.py.
DB_CONNECT_RETRY_BUDGET_S = 35


def _connect_with_resume_retry(**kwargs):
deadline = time.monotonic() + DB_CONNECT_RETRY_BUDGET_S
delay, logged = 1, False
while True:
try:
return psycopg2.connect(**kwargs)
except psycopg2.OperationalError as e:
if "authentication failed" in str(e) or time.monotonic() + delay > deadline:
raise
if not logged:
print(f"[db] connect failed, database may be resuming from auto-pause; retrying "
f"for up to {DB_CONNECT_RETRY_BUDGET_S}s: {str(e).strip()[:200]}")
logged = True
time.sleep(delay)
delay = min(delay * 2, 8)


def get_db_conn():
from urllib.parse import urlparse
url = os.environ["DATABASE_URL"]
p = urlparse(url)
host, port, user, dbname = p.hostname, p.port or 5432, p.username, p.path.lstrip("/")
rds = boto3.client("rds", region_name=os.environ.get("AWS_REGION", "us-east-1"))
token = rds.generate_db_auth_token(DBHostname=host, Port=port, DBUsername=user)
return psycopg2.connect(host=host, port=port, user=user, password=token, dbname=dbname, sslmode="require")
return _connect_with_resume_retry(host=host, port=port, user=user, password=token, dbname=dbname, sslmode="require")


def detect_format(key: str, content_type: str, file_bytes: bytes = b"") -> str:
Expand Down
83 changes: 83 additions & 0 deletions lambdas/tests/test_db_resume_retry.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
"""Every Lambda that connects to Aurora retries a failed connect while the cluster resumes
from auto-pause, never retries authentication errors, and gives up after its budget."""
import psycopg2
import pytest


@pytest.fixture(params=["profiler", "executor", "drift"])
def handler(request):
return request.getfixturevalue(request.param)


class FakeClock:
def __init__(self):
self.t = 0.0
self.sleeps = []

def monotonic(self):
return self.t

def sleep(self, s):
self.sleeps.append(s)
self.t += s


@pytest.fixture
def clock(handler, monkeypatch):
c = FakeClock()
monkeypatch.setattr(handler.time, "monotonic", c.monotonic)
monkeypatch.setattr(handler.time, "sleep", c.sleep)
return c


def _connect_failing(monkeypatch, handler, errors, result="conn"):
calls = []

def fake_connect(**kwargs):
calls.append(kwargs)
if len(calls) <= len(errors):
raise errors[len(calls) - 1]
return result

monkeypatch.setattr(handler.psycopg2, "connect", fake_connect)
return calls


def test_retries_while_resuming_then_connects(handler, clock, monkeypatch, capsys):
err = psycopg2.OperationalError("could not connect to server: Connection timed out")
calls = _connect_failing(monkeypatch, handler, [err, err, err])
assert handler._connect_with_resume_retry(host="h", dbname="d") == "conn"
assert len(calls) == 4 and calls[-1] == {"host": "h", "dbname": "d"}
assert clock.sleeps == [1, 2, 4]
out = capsys.readouterr().out
assert out.count("database may be resuming") == 1 # one log line per connect


def test_authentication_errors_are_not_retried(handler, clock, monkeypatch):
err = psycopg2.OperationalError('FATAL: PAM authentication failed for user "cleanstack"')
calls = _connect_failing(monkeypatch, handler, [err])
with pytest.raises(psycopg2.OperationalError, match="authentication failed"):
handler._connect_with_resume_retry(host="h")
assert len(calls) == 1 and clock.sleeps == []


def test_other_exceptions_are_not_retried(handler, clock, monkeypatch):
calls = _connect_failing(monkeypatch, handler, [ValueError("bad dsn")])
with pytest.raises(ValueError):
handler._connect_with_resume_retry(host="h")
assert len(calls) == 1 and clock.sleeps == []


def test_gives_up_after_the_budget(handler, clock, monkeypatch):
err = psycopg2.OperationalError("could not connect to server")
calls = _connect_failing(monkeypatch, handler, [err] * 100)
with pytest.raises(psycopg2.OperationalError):
handler._connect_with_resume_retry(host="h")
assert clock.sleeps == [1, 2, 4, 8, 8, 8]
assert sum(clock.sleeps) <= handler.DB_CONNECT_RETRY_BUDGET_S
assert len(calls) == len(clock.sleeps) + 1


def test_get_db_conn_uses_the_retry(handler):
import inspect
assert "_connect_with_resume_retry(" in inspect.getsource(handler.get_db_conn)
Loading
Loading