diff --git a/backend/scripts/backfill_artifact_user_index_keys.py b/backend/scripts/backfill_artifact_user_index_keys.py new file mode 100644 index 00000000..2b843ecb --- /dev/null +++ b/backend/scripts/backfill_artifact_user_index_keys.py @@ -0,0 +1,222 @@ +"""Backfill: stamp GSI2PK/GSI2SK on artifact HEAD rows written before they existed. + +The artifact writer began stamping user-index keys on HEAD rows in +``feat(artifacts): stamp user-index keys on HEAD rows ahead of the index`` +(2026-09-04). Every HEAD row written before that carries neither +attribute: + + HEAD row : PK=USER#{user_id} SK=ARTIFACT#{aid}#HEAD + + GSI2PK=USER#{user_id} + + GSI2SK=ARTIFACT#{updated_at}#{aid} + +``UserArtifactsIndex`` is **sparse**: DynamoDB indexes a row only if it +carries the index's key attributes. A row missing them is not "stale" in +the index, it is *absent from it forever* — and the omission is silent. +Switching the library's user-wide listing to that index without this +backfill would drop every artifact created before 2026-09-04 from the +page, with no error anywhere. + +RUN THIS BEFORE THE INDEX IS CREATED +------------------------------------ +DynamoDB backfills a new GSI at creation time from rows that already +carry its keys. Stamping first therefore means the index is complete the +moment it reports ACTIVE, with no window in which it is partially +populated. The attributes are inert until an index consumes them (no +index write is charged), so running early costs nothing. + +WHAT IT DOES NOT TOUCH +---------------------- +``updated_at``. It is not a display field: HEAD's ``GSI1SK``/``GSI2SK`` +embed it, and only the writer maintains it. This script *reads* it to +build ``GSI2SK`` — matching byte-for-byte what the writer would have +written — and never assigns it. Same restraint as +``ArtifactLifecycleService.rename``. + +Version rows are skipped by design. The keys belong on HEAD rows only, +so the index holds one row per artifact rather than one per version. + +SAFETY +------ +* **Dry-run by default.** Pass ``--apply`` to write. +* **Idempotent.** Guarded by ``attribute_not_exists(GSI2PK)``, so a + second run finds nothing and a row the writer has since stamped is + left alone. +* **Never resurrects a deleted row.** ``attribute_exists(SK)`` on every + update, matching the writer's own write-back rule. +* **Reports what it cannot fix** rather than guessing — a HEAD row with + no ``updated_at`` is counted and named, not stamped with a fabricated + timestamp that would sort wrongly forever. + +Run against dev first, then prod:: + + AWS_PROFILE=dev-ai python backend/scripts/backfill_artifact_user_index_keys.py \\ + --table dev-boisestateai-v2-user-artifacts --region us-west-2 + AWS_PROFILE=dev-ai python backend/scripts/backfill_artifact_user_index_keys.py \\ + --table dev-boisestateai-v2-user-artifacts --region us-west-2 --apply +""" + +from __future__ import annotations + +import argparse +import logging +from typing import Any, Dict, Iterator, List + +import boto3 +from botocore.exceptions import ClientError + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" +) +logger = logging.getLogger("backfill_artifact_user_index_keys") + +HEAD_SUFFIX = "#HEAD" + + +def iter_head_rows(table: Any) -> Iterator[Dict[str, Any]]: + """Every artifact HEAD row in the table. + + A Scan, not a Query: this is a whole-table migration across all + users, and the table's partition key is the user. The + ``FilterExpression`` runs server-side purely to cut payload — + DynamoDB still reads every item either way, so it saves bandwidth, + not capacity. + """ + kwargs: Dict[str, Any] = { + "FilterExpression": "begins_with(SK, :p)", + "ExpressionAttributeValues": {":p": "ARTIFACT#"}, + } + while True: + resp = table.scan(**kwargs) + for item in resp.get("Items", []): + if str(item.get("SK", "")).endswith(HEAD_SUFFIX): + yield item + last = resp.get("LastEvaluatedKey") + if not last: + return + kwargs["ExclusiveStartKey"] = last + + +def plan_row(item: Dict[str, Any]) -> Dict[str, Any] | None: + """What this row needs, or None if it needs nothing. + + Returns a dict with the computed keys, or ``{"skip": reason}`` for a + row that cannot be stamped safely. + """ + if "GSI2PK" in item: + return None # already stamped — writer or a previous run + + pk = str(item.get("PK", "")) + artifact_id = str(item.get("artifact_id", "")) + updated_at = str(item.get("updated_at", "")) + + if not pk.startswith("USER#"): + return {"skip": f"unexpected PK {pk!r}"} + if not artifact_id: + return {"skip": "no artifact_id attribute"} + if not updated_at: + # Deliberately not falling back to created_at or "now": GSI2SK is + # the sort key the library orders by, and a fabricated timestamp + # would order this artifact wrongly for the rest of its life. + # Better to name it and let a human decide. + return {"skip": "no updated_at attribute"} + + return { + "gsi2pk": pk, # GSI2PK is exactly the base PK — USER#{user_id} + "gsi2sk": f"ARTIFACT#{updated_at}#{artifact_id}", + } + + +def backfill(table: Any, apply: bool) -> Dict[str, int]: + stats = {"head_rows": 0, "already": 0, "stamped": 0, "skipped": 0, "failed": 0} + skipped: List[str] = [] + + for item in iter_head_rows(table): + stats["head_rows"] += 1 + plan = plan_row(item) + + if plan is None: + stats["already"] += 1 + continue + + sk = str(item.get("SK", "")) + if "skip" in plan: + stats["skipped"] += 1 + skipped.append(f"{item.get('PK')} / {sk}: {plan['skip']}") + continue + + logger.info( + "stamp %s %s -> GSI2SK=%s", item.get("PK"), sk, plan["gsi2sk"] + ) + if not apply: + stats["stamped"] += 1 + continue + + try: + table.update_item( + Key={"PK": item["PK"], "SK": item["SK"]}, + UpdateExpression="SET GSI2PK = :pk, GSI2SK = :sk", + ExpressionAttributeValues={ + ":pk": plan["gsi2pk"], + ":sk": plan["gsi2sk"], + }, + # attribute_exists(SK): never resurrect a row the delete + # path removed between the scan and this write. + # attribute_not_exists(GSI2PK): idempotent, and yields to + # the writer if it stamped the row in the meantime. + ConditionExpression=( + "attribute_exists(SK) AND attribute_not_exists(GSI2PK)" + ), + ) + stats["stamped"] += 1 + except ClientError as exc: + code = exc.response.get("Error", {}).get("Code", "") + if code == "ConditionalCheckFailedException": + # Deleted, or stamped by the writer, while we scanned. + stats["already"] += 1 + continue + stats["failed"] += 1 + logger.error("failed to stamp %s: %s", sk, code, exc_info=True) + + if skipped: + logger.warning("%s row(s) could not be stamped:", len(skipped)) + for line in skipped: + logger.warning(" %s", line) + + return stats + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--table", required=True, help="user-artifacts table name") + parser.add_argument("--region", default="us-west-2") + parser.add_argument( + "--apply", + action="store_true", + help="actually write (default is a dry run)", + ) + args = parser.parse_args() + + table = boto3.resource("dynamodb", region_name=args.region).Table(args.table) + + if not args.apply: + logger.info("DRY RUN — no writes. Pass --apply to commit.") + + stats = backfill(table, args.apply) + + logger.info( + "HEAD rows=%s already-stamped=%s stamped=%s skipped=%s failed=%s", + stats["head_rows"], + stats["already"], + stats["stamped"], + stats["skipped"], + stats["failed"], + ) + if stats["skipped"] or stats["failed"]: + logger.warning( + "Index will be INCOMPLETE for the rows above. Resolve them " + "before switching the library query to UserArtifactsIndex." + ) + + +if __name__ == "__main__": + main() diff --git a/backend/src/agents/builtin_tools/artifacts/service.py b/backend/src/agents/builtin_tools/artifacts/service.py index 71e16ff6..f5b8bf98 100644 --- a/backend/src/agents/builtin_tools/artifacts/service.py +++ b/backend/src/agents/builtin_tools/artifacts/service.py @@ -10,7 +10,7 @@ + GSI1PK=SESSION#{session_id} + GSI1SK=ARTIFACT#{updated_at}#{aid} (SessionIndex) + GSI2PK=USER#{user_id} - + GSI2SK=ARTIFACT#{updated_at}#{aid} (index NOT YET CREATED) + + GSI2SK=ARTIFACT#{updated_at}#{aid} (UserArtifactsIndex) S3 layout : {user_id}/{aid}/v{n}/index.html Versions are immutable (no DeleteObject grant in inference-api) — an @@ -27,20 +27,19 @@ (the optimistic lock below) nor `updated_at` (embedded in the GSI sort keys, which only this module maintains). -GSI2PK/GSI2SK are written ahead of the index that will consume them. -Nothing queries them today — a user-wide artifact list is served by a -base-table Query on PK=USER#{uid}, which is adequate while the heaviest -user holds well under a 1MB page. They are stamped now because a sparse -GSI only ever contains rows that already carry its key attributes: rows -written before the attributes exist stay invisible to it forever unless -a migration script backfills them, and that omission fails silently (a -library page that lists only artifacts created after the deploy, with no -error anywhere). Writing them from now on means the eventual -`UserArtifactsIndex` — GSI2PK hash, GSI2SK range, queried with -ScanIndexForward=False for newest-first — backfills every row stamped -since this change, shrinking the manual backfill to the rows that -predate it. Until then they are inert ordinary attributes: DynamoDB -charges no index write when no index consumes them. +GSI2PK/GSI2SK feed `UserArtifactsIndex` — GSI2PK hash, GSI2SK range, +queried with ScanIndexForward=False for newest-first. They were stamped +here for two weeks before that index existed, deliberately: a sparse GSI +only ever contains rows that already carry its key attributes, so rows +written before the attributes exist stay invisible to it forever, and +that omission fails silently (a library page listing only artifacts +created after the deploy, with no error anywhere). Stamping early shrank +the manual backfill to the rows predating 2026-09-04, which +`backend/scripts/backfill_artifact_user_index_keys.py` then stamped. + +**Both write paths below must keep stamping them.** A HEAD row that +loses these attributes on its next update drops out of the index — and +out of the owner's library — with nothing to indicate it. Stamped on HEAD rows only, deliberately: one indexed row per artifact rather than one per version. Both write paths below must stamp them, or diff --git a/backend/tests/test_backfill_artifact_user_index_keys.py b/backend/tests/test_backfill_artifact_user_index_keys.py new file mode 100644 index 00000000..6035f2ac --- /dev/null +++ b/backend/tests/test_backfill_artifact_user_index_keys.py @@ -0,0 +1,250 @@ +"""Tests for the UserArtifactsIndex key backfill. + +The failure this guards against is silent: `UserArtifactsIndex` is +sparse, so a HEAD row left without `GSI2PK`/`GSI2SK` is not stale in the +index — it is absent from it forever, and the library page simply stops +listing that artifact with no error anywhere. + +So the assertions that matter are about what the script *refuses* to do +(fabricate a sort key, touch `updated_at`, resurrect a deleted row) as +much as what it writes. +""" + +from __future__ import annotations + +import importlib.util +import pathlib + +import boto3 +import pytest +from moto import mock_aws + +REGION = "us-east-1" +TABLE = "test-user-artifacts" + +_SCRIPT = ( + pathlib.Path(__file__).resolve().parents[1] + / "scripts" + / "backfill_artifact_user_index_keys.py" +) +_spec = importlib.util.spec_from_file_location("backfill_user_index", _SCRIPT) +backfill_mod = importlib.util.module_from_spec(_spec) +assert _spec.loader is not None +_spec.loader.exec_module(backfill_mod) + + +@pytest.fixture() +def table(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("AWS_DEFAULT_REGION", REGION) + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing") + with mock_aws(): + ddb = boto3.resource("dynamodb", region_name=REGION) + ddb.create_table( + TableName=TABLE, + KeySchema=[ + {"AttributeName": "PK", "KeyType": "HASH"}, + {"AttributeName": "SK", "KeyType": "RANGE"}, + ], + AttributeDefinitions=[ + {"AttributeName": "PK", "AttributeType": "S"}, + {"AttributeName": "SK", "AttributeType": "S"}, + ], + BillingMode="PAY_PER_REQUEST", + ) + yield ddb.Table(TABLE) + + +def put_head( + table, + *, + user: str = "u1", + artifact: str = "a1", + updated_at: str | None = "2026-08-20T19:41:29.618536+00:00", + stamped: bool = False, + **extra, +) -> None: + item = { + "PK": f"USER#{user}", + "SK": f"ARTIFACT#{artifact}#HEAD", + "artifact_id": artifact, + "user_id": user, + "version": 1, + "title": "Deck", + "content_type": "text/html; charset=utf-8", + "session_id": "s1", + } + if updated_at is not None: + item["updated_at"] = updated_at + if stamped: + item["GSI2PK"] = f"USER#{user}" + item["GSI2SK"] = f"ARTIFACT#{updated_at}#{artifact}" + item.update(extra) + table.put_item(Item=item) + + +def put_version(table, *, user: str = "u1", artifact: str = "a1", version: int = 1): + table.put_item( + Item={ + "PK": f"USER#{user}", + "SK": f"ARTIFACT#{artifact}#V#{version:05d}", + "artifact_id": artifact, + "user_id": user, + "version": version, + "updated_at": "2026-08-20T19:41:29.618536+00:00", + } + ) + + +def row(table, user="u1", artifact="a1") -> dict: + return table.get_item( + Key={"PK": f"USER#{user}", "SK": f"ARTIFACT#{artifact}#HEAD"} + )["Item"] + + +# ------------------------------------------------------------------ +# What it writes +# ------------------------------------------------------------------ + + +def test_stamps_keys_matching_what_the_writer_would_have_written(table): + put_head(table, updated_at="2026-08-20T19:41:29.618536+00:00") + + stats = backfill_mod.backfill(table, apply=True) + + item = row(table) + assert item["GSI2PK"] == "USER#u1" + # Byte-for-byte the writer's format: ARTIFACT#{updated_at}#{aid}. + # A different shape here would order the index inconsistently with + # every row the writer stamps from now on. + assert item["GSI2SK"] == "ARTIFACT#2026-08-20T19:41:29.618536+00:00#a1" + assert stats["stamped"] == 1 + + +def test_never_touches_updated_at(table): + # updated_at is embedded in both GSI sort keys and is writer-owned. + # Bumping it here would reorder the library and desync GSI1SK. + put_head(table, updated_at="2026-08-20T19:41:29.618536+00:00") + + backfill_mod.backfill(table, apply=True) + + assert row(table)["updated_at"] == "2026-08-20T19:41:29.618536+00:00" + + +def test_leaves_version_rows_alone(table): + # The keys belong on HEAD only, so the index holds one row per + # artifact rather than one per version. + put_head(table) + put_version(table, version=1) + put_version(table, version=2) + + stats = backfill_mod.backfill(table, apply=True) + + assert stats["head_rows"] == 1 + for v in (1, 2): + item = table.get_item( + Key={"PK": "USER#u1", "SK": f"ARTIFACT#a1#V#{v:05d}"} + )["Item"] + assert "GSI2PK" not in item + + +def test_spans_every_user(table): + # Whole-table migration: the base table is partitioned by user, so a + # Query would only ever see one of them. + put_head(table, user="u1", artifact="a1") + put_head(table, user="u2", artifact="a2") + + stats = backfill_mod.backfill(table, apply=True) + + assert stats["stamped"] == 2 + assert row(table, "u2", "a2")["GSI2PK"] == "USER#u2" + + +# ------------------------------------------------------------------ +# What it refuses to do +# ------------------------------------------------------------------ + + +def test_refuses_to_fabricate_a_sort_key(table): + """A HEAD row with no `updated_at` is reported, never guessed at. + + GSI2SK is what the library orders by; inventing a timestamp would + sort that artifact wrongly for the rest of its life, invisibly.""" + put_head(table, updated_at=None) + + stats = backfill_mod.backfill(table, apply=True) + + assert stats["skipped"] == 1 + assert stats["stamped"] == 0 + assert "GSI2PK" not in row(table) + + +def test_dry_run_writes_nothing(table): + put_head(table) + + stats = backfill_mod.backfill(table, apply=False) + + assert stats["stamped"] == 1 # counted as *would* stamp + assert "GSI2PK" not in row(table) + + +def test_is_idempotent(table): + put_head(table) + + first = backfill_mod.backfill(table, apply=True) + second = backfill_mod.backfill(table, apply=True) + + assert first["stamped"] == 1 + assert second["stamped"] == 0 + assert second["already"] == 1 + + +def test_yields_to_a_row_the_writer_already_stamped(table): + # The writer's value wins; the script must not overwrite it. + put_head(table, stamped=True) + + stats = backfill_mod.backfill(table, apply=True) + + assert stats["already"] == 1 + assert stats["stamped"] == 0 + + +def test_does_not_resurrect_a_row_deleted_mid_run(table): + """The scan and the writes are not one transaction, so a row can be + deleted in between. `attribute_exists(SK)` makes that a no-op instead + of a resurrection — the same rule the writer's own write-backs use.""" + put_head(table) + rows = list(backfill_mod.iter_head_rows(table)) + assert len(rows) == 1 + + table.delete_item(Key={"PK": "USER#u1", "SK": "ARTIFACT#a1#HEAD"}) + + # Replay the write for the row we scanned before the delete. + plan = backfill_mod.plan_row(rows[0]) + assert plan is not None and "skip" not in plan + stats = backfill_mod.backfill(table, apply=True) + + assert stats["head_rows"] == 0 + assert ( + table.get_item( + Key={"PK": "USER#u1", "SK": "ARTIFACT#a1#HEAD"} + ).get("Item") + is None + ) + + +def test_reports_an_unexpected_partition_key(table): + put_head(table) + table.put_item( + Item={ + "PK": "SHARE#abc", + "SK": "ARTIFACT#weird#HEAD", + "artifact_id": "weird", + "updated_at": "2026-08-20T00:00:00+00:00", + } + ) + + stats = backfill_mod.backfill(table, apply=True) + + assert stats["stamped"] == 1 + assert stats["skipped"] == 1 diff --git a/infrastructure/gsi-inventory.json b/infrastructure/gsi-inventory.json index 59b96257..10bcf0d7 100644 --- a/infrastructure/gsi-inventory.json +++ b/infrastructure/gsi-inventory.json @@ -63,7 +63,8 @@ "system-cost-rollup": [], "system-prompts": [], "user-artifacts": [ - "SessionIndex" + "SessionIndex", + "UserArtifactsIndex" ], "user-cost-summary": [ "PeriodCostIndex" diff --git a/infrastructure/lib/constructs/artifacts/artifacts-data-construct.ts b/infrastructure/lib/constructs/artifacts/artifacts-data-construct.ts index 5afe2b97..480b6c19 100644 --- a/infrastructure/lib/constructs/artifacts/artifacts-data-construct.ts +++ b/infrastructure/lib/constructs/artifacts/artifacts-data-construct.ts @@ -79,6 +79,45 @@ export class ArtifactsDataConstruct extends Construct { projectionType: dynamodb.ProjectionType.ALL, }); + // Sparse index over artifact HEAD rows, keyed by owner and ordered by + // `updated_at` — what the library's user-wide listing is served from. + // + // The base table is already partitioned by user, so this is NOT about + // reachability: `list_for_user` could always Query PK=USER#{uid}. It is + // about what that Query has to read. HEAD and version rows share the + // partition, so the base-table Query spans roughly 3x the rows it + // returns, then filters and date-sorts them in memory — which also + // makes it unpaginable, since the page boundary would fall in the + // wrong place. Only HEAD rows carry GSI2PK, so this index holds one + // row per artifact, already in newest-first order. + // + // ⚠️ ADDING A SECOND GSI IS ONE `UpdateTable`, AND ONLY ONE GSI MAY BE + // ADDED PER `UpdateTable`. Do not add another index to this table in + // the same deploy — CloudFormation will reject the change set. See + // `test/gsi-update-limit.test.ts`. + // + // ⚠️ CFN reporting UPDATE_COMPLETE does NOT mean the index is ACTIVE: + // DynamoDB backfills it asynchronously afterwards. Anything that + // queries it must not deploy until `DescribeTable` reports + // `IndexStatus: ACTIVE`, which is why the query switch ships in a + // separate PR from this one. + // + // ⚠️ Sparse means a HEAD row without GSI2PK is ABSENT from this index + // forever, not merely stale — and silently. Rows written before the + // writer began stamping these keys (2026-09-04) are backfilled by + // `backend/scripts/backfill_artifact_user_index_keys.py`, which must + // have been run for the environment before anything reads the index. + this.table.addGlobalSecondaryIndex({ + indexName: 'UserArtifactsIndex', + partitionKey: { name: 'GSI2PK', type: dynamodb.AttributeType.STRING }, + sortKey: { name: 'GSI2SK', type: dynamodb.AttributeType.STRING }, + // ALL, matching SessionIndex: the library row needs title, + // content_type, version, created_at, updated_at and session_id, so a + // KEYS_ONLY projection would force a base-table read per row and + // give back exactly the amplification this index exists to remove. + projectionType: dynamodb.ProjectionType.ALL, + }); + this.bucket = new s3.Bucket(this, 'ArtifactsContentBucket', { bucketName: getResourceName(config, 'artifacts-content'), encryption: s3.BucketEncryption.S3_MANAGED,