Setting the file. One moment.
Benchmark Lambda · Amazon DynamoDB · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Reference Architecture
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def __init__
— line 237
This file
Number 65.7
Position 7 of 14
Type Python
Size 78 KB
Lines 1,724 scripts/ benchmark_lambda.py
Python · 1,724 lines · 78 KB
16 {
17 "phase_plan": ["settle","seed","warmup","measure"] | ["measure"],
18 "invocation_index": 0, # 0 for the first call, then 1, 2, ...
19 "invocations_total": 1, # how many invocations for this run
20 "patterns": [ <access_pattern>, ... ], # from design JSON
21 "tables": [ <table_def>, ... ], # from design JSON (includes key_schema)
22 "manifest": { deploy_model.py output — tables[].name (prefixed) },
23 "config": { ...benchmark_config.json knobs... }
24 }
25
26 Response schema:
27
28 {
29 "invocation_index": 0,
30 "phases_run": ["settle","seed","warmup","measure"],
31 "raw_rows": [ {pattern_id, op, phase, ts, latency_ms, consumed_cu,
32 gsi_cu, throttled, error}, ... ],
33 "seed_verification": { table: {expected, actual}, ... },
34 "coverage": { "measured_patterns": [...], "missing_patterns": [...],
35 "coverage_incomplete": bool },
36 "measurement_tainted": { pattern_id: "warmup"|"ramp"|... },
37 "lambda_duration_seconds": float
38 }
39
40 Raw-row payload is kept under the 6MB Lambda response limit by capping rows
41 per invocation; when measurement spans multiple invocations the orchestrator
42 concatenates across responses.
43 """
44
45 from __future__ import annotations
46
47 import bisect
48 import os
49 import random
50 import threading
51 import time
52 from concurrent.futures import ThreadPoolExecutor
53 from typing import Any
54
55 import boto3
56 from botocore.config import Config as BotoConfig
57 from botocore.exceptions import ClientError
58
59 READ_OPS = { "GetItem" , "Query" , "Scan" , "BatchGetItem" , "TransactGetItems" }
60 # SearchVectors is a read but consumes NO RCU — it is metered in bytes
61 # (VectorSearchRequestBytes), so it is tracked separately from consumed_cu rather than
62 # folded into READ_OPS, where a naive report would show it as a free operation.
63 VECTOR_SEARCH_OP = "SearchVectors"
64 WRITE_OPS = { "PutItem" , "UpdateItem" , "DeleteItem" , "BatchWriteItem" , "TransactWriteItems" }
65 THROTTLE_CODES = (
66 "ProvisionedThroughputExceededException" ,
67 "ThrottlingException" ,
68 "RequestLimitExceeded" ,
69 )
70
71 # Keep Lambda response under the 6MB payload limit. Each row is roughly
72 # 200-300 bytes JSON-encoded; 15k leaves headroom for gsi_cu dicts and errors.
73 # This is the HARD payload safety bound — the sum of PHASE_ROW_BUDGET below is
74 # kept at or under it so the per-phase reservation never exceeds the payload.
75 MAX_ROWS_PER_INVOCATION = 15_000
76
77 # Per-PHASE row budget. A single shared global cap is wrong: warmup runs for
78 # EVERY pattern before any measure row is recorded, so on a multi-pattern run
79 # that fits in one invocation, warmup fills the whole 15k budget and `measure`
80 # gets ZERO latency rows — the p50/p99 columns (and the hot-vs-cold p99 signal
81 # that drives key_skew on on-demand tables) come back empty. Reserving a fixed
82 # budget per phase guarantees `measure` always keeps its allocation no matter
83 # how many warmup rows were produced. settle/seed are tiny in practice; the bulk
84 # goes to measure. The four budgets sum to MAX_ROWS_PER_INVOCATION so the hard
85 # payload bound still holds. Exact call/throttle COUNTS are tracked separately
86 # (uncapped) in run_pattern_window, so down-sampling recorded ROWS only thins the
87 # latency percentiles, never the throttle tally.
88 PHASE_ROW_BUDGET = {
89 "settle" : 500 ,
90 "seed" : 1_500 ,
91 "warmup" : 4_000 ,
92 "measure" : 9_000 ,
93 }
94 # Floor for the per-(pattern, phase) sub-cap so a design with many patterns
95 # still records a usable per-pattern latency sample.
96 MIN_ROWS_PER_PATTERN_PHASE = 200
97
98 # Seeded-key namespace is deterministic per pattern so a second invocation
99 # can resume without a shared manifest.
100 #
101 # Key VALUES must match the key attribute's declared DDB type. A string key gets
102 # the readable "bench#<pattern>#pk<idx>" form. A NUMERIC key (type "N") cannot
103 # carry a string prefix, so we encode the pattern + role + index into a single
104 # deterministic integer: a per-(pattern,role) "bank" offset (hash of the label,
105 # kept well inside JS/DDB safe-integer range) plus the index. Distinct patterns,
106 # distinct roles (pk vs sk), and distinct indices therefore never collide — the
107 # same uniqueness guarantee the string form gives, which the batch/transact
108 # distinct-key walk depends on. A binary key ("B") gets the UTF-8 bytes of the
109 # string form. Unknown/defaulted type is "S" (historical behavior).
110 # Indices per (pattern,role) numeric "bank". Cross-bank disjointness — and thus
111 # the collision-freedom the batch/transact distinct-key walk relies on — holds
112 # as long as the per-bank index stays below this stride. Seed indices are
113 # bounded by seed_items_per_table (default 500; thousands at most), so the 10M
114 # headroom is never approached in practice.
115 _NUM_BANK_STRIDE = 10_000_000
116
117
118 def _bank_offset (pattern_id: str , role: str ) -> int :
119 # Stable, process-independent offset in [0, ~9e15) — comfortably within DDB's
120 # 38-digit number range and JS safe-int (2^53) so no precision is lost.
121 h = 0
122 for ch in f " { pattern_id } # { role } " :
123 h = (h * 131 + ord (ch)) & 0x FFFFFFFF
124 return (h % 900_000_000 ) * _NUM_BANK_STRIDE
125
126
127 def _seed_key_val (pattern_id: str , idx: int , role: str , ktype: str ):
128 """Deterministic, type-correct, collision-free key value.
129
130 role is "pk" or "sk"; ktype is the declared DDB type ("S"/"N"/"B")."""
131 if ktype == "N" :
132 return _bank_offset(pattern_id, role) + idx
133 s = f "bench# { pattern_id } # { role }{ idx :06d} "
134 if ktype == "B" :
135 return s.encode( "utf-8" )
136 return s
137
138
139 # GSI synthetic key value — type-aware, and IDENTICAL between the seed side and
140 # the query side so a GSI Query finds the items the seed wrote. role is "pk"/"sk".
141 def _gsi_val (pattern_id: str , idx: int , role: str , ktype: str ):
142 if ktype == "N" :
143 return _bank_offset( f "gsi# { pattern_id } " , role) + idx
144 s = f "bench#gsi- { role } # { pattern_id } # { idx } "
145 if ktype == "B" :
146 return s.encode( "utf-8" )
147 return s
148
149
150 # Back-compat string generators (retained for any remaining string-only callers).
151 def _seed_pk (pattern_id: str , idx: int ) -> str :
152 return f "bench# { pattern_id } #pk { idx :06d} "
153
154
155 def _seed_sk (pattern_id: str , idx: int ) -> str :
156 return f "bench# { pattern_id } #sk { idx :06d} "
157
158
159 # ---------------------------------------------------------------------------
160 # DDB type helpers (low-level client uses type-annotated JSON)
161 # ---------------------------------------------------------------------------
162
163
164 def _serialize (value):
165 """Minimal DDB low-level serializer for strings/numbers/bytes/bool/null."""
166 if isinstance (value, str ):
167 return { "S" : value}
168 if isinstance (value, bool ):
169 return { "BOOL" : value}
170 if isinstance (value, ( int , float )):
171 return { "N" : str (value)}
172 if value is None :
173 return { "NULL" : True }
174 if isinstance (value, bytes ):
175 return { "B" : value}
176 if isinstance (value, ( list , tuple )):
177 # A vector attribute. Must be an L of N — see _serialize_vector.
178 return _serialize_vector(value)
179 raise ValueError ( f "unsupported attribute type for { value !r} " )
180
181
182 def _serialize_vector (values):
183 """A vector attribute is a DynamoDB List of Numbers — never a Number Set.
184
185 Sending NS is rejected with the misleading 'Input collection contains duplicates'
186 (the set collapses repeated values), which is why this is explicit.
187 """
188 return { "L" : [{ "N" : f " { float (v) :.6f} " } for v in values]}
189
190
191 def _build_item (pk_attr, pk_val, sk_attr, sk_val, target_size_bytes, gsi_attrs = None , extra = None ):
192 """Build a DDB item padded to target_size_bytes using a blob attribute."""
193 item = {pk_attr: _serialize(pk_val)}
194 if sk_attr and sk_val is not None :
195 item[sk_attr] = _serialize(sk_val)
196 for name, val in (gsi_attrs or {}).items():
197 item[name] = _serialize(val)
198 for name, val in (extra or {}).items():
199 item[name] = _serialize(val)
200
201 # Pad using a single "payload" String attribute. Account for the small
202 # attribute-name overhead; target is approximate.
203 def _attr_value_bytes (v):
204 # A vector attribute is an L of N, and it MUST be counted. Missing it is not a
205 # rounding error: a 1024-dim embedding is ~8.7 KB, so an item declared at 2,048 B
206 # was being padded as though the vector weren't there and landed at 10,745 B —
207 # 5.2x the declared size, which inflates base-table WCU for exactly the vector
208 # designs this benchmark is meant to measure. Recursive because L can nest.
209 if "L" in v:
210 return sum (_attr_value_bytes(e) for e in v[ "L" ])
211 if "M" in v:
212 return sum ( len (k) + _attr_value_bytes(e) for k, e in v[ "M" ].items())
213 for t in ( "S" , "N" , "B" ):
214 if t in v:
215 return len (v[t])
216 # BOOL/NULL carry no value bytes and count as their key length only. That
217 # under-counts by a few bytes, which pads VERY slightly larger — never smaller —
218 # so the item still meets target_size_bytes.
219 return 0
220
221 def _size (it):
222 return sum ( len (k) + _attr_value_bytes(v) for k, v in it.items())
223
224 current = _size(item)
225 if target_size_bytes and current < target_size_bytes:
226 pad_len = max ( 1 , target_size_bytes - current - len ( "payload" ))
227 item[ "payload" ] = { "S" : "x" * pad_len}
228 return item
229
230
231 # ---------------------------------------------------------------------------
232 # Context: resolves table specs, attribute names, prefixed table names
233 # ---------------------------------------------------------------------------
234
235
236 class RunContext :
237 def __init__ (self, event, client):
238 self .cfg = event.get( "config" ) or {}
239 self .patterns = event.get( "patterns" ) or []
240 self .tables = event.get( "tables" ) or []
241 self .manifest = event.get( "manifest" ) or {}
242 self .client = client
243 self .phase_plan = event.get( "phase_plan" ) or [ "settle" , "seed" , "warmup" , "measure" ]
244 self .invocation_index = int (event.get( "invocation_index" , 0 ))
245 self .invocations_total = int (event.get( "invocations_total" , 1 ))
246 self .run_id = self .manifest.get( "run_id" ) or "unknown"
247
248 # Map original table name -> prefixed table name.
249 self .prefixed = {}
250 for t in self .manifest.get( "tables" ) or []:
251 orig = t.get( "original_name" ) or t[ "name" ]
252 self .prefixed[orig] = t[ "name" ]
253
254 # Map table name -> table def (with key_schema, gsis, entities).
255 self .table_by_name = {t[ "table_name" ]: t for t in self .tables}
256
257 # Map (table_name, attribute_name) -> declared DDB scalar type ("S"/"N"/
258 # "B"). Key generation MUST honor this: a key attribute declared "N"
259 # (e.g. a numeric `recorded_at` or epoch `order_date` sort key) rejects a
260 # string value with ValidationException, so the seed/read/write key value
261 # has to match the declared type. Defaults to "S" when a type isn't
262 # declared (the common case and the historical behavior), so string-keyed
263 # designs are unaffected.
264 #
265 # Types are read from TWO sources, in increasing precedence:
266 # 1. entities[].attributes[] as {"name","type"} — the canonical schema
267 # form documented in references/cost-model-schema.md.
268 # 2. a table-level "attribute_definitions" block — the raw-CreateTable
269 # -API spelling an author (or LLM) naturally reaches for. Accepts
270 # BOTH {"attribute_name","attribute_type"} (API style) and the
271 # {"name","type"} shorthand. An explicit attribute_definitions entry
272 # WINS over an entities-derived type for the same attribute.
273 #
274 # CRITICAL: scripts/deploy_model.py (_collect_attr_types) parses these
275 # exact two sources with the same precedence. If the two ever diverge —
276 # deploy creates order_date as N but this map thinks it's S — key
277 # generation emits the wrong type and every write fails with
278 # ValidationException (the W4 bug). Keep them in sync.
279 def _norm_t (v):
280 v = (v or "S" ).upper()
281 return v if v in ( "S" , "N" , "B" ) else "S"
282
283 self .attr_types: dict[ tuple , str ] = {}
284 for t in self .tables:
285 tn = t[ "table_name" ]
286 # 1. entities[].attributes[] (lower precedence)
287 for ent in t.get( "entities" ) or []:
288 for a in ent.get( "attributes" ) or []:
289 name = a.get( "name" )
290 if name:
291 self .attr_types[(tn, name)] = _norm_t(a.get( "type" ))
292 # 2. table-level attribute_definitions (higher precedence)
293 for a in t.get( "attribute_definitions" ) or []:
294 name = a.get( "attribute_name" ) or a.get( "name" )
295 if name:
296 self .attr_types[(tn, name)] = _norm_t(a.get( "attribute_type" ) or a.get( "type" ))
297
298 # Seed items per table — the number written in the seed phase. This is
299 # also the size of the distinct-key pool the read-key sampler draws over
300 # (see n_partitions below and n_distinct_keys in _dispatch); there is no
301 # separate key-space knob.
302 self .seed_items = int ( self .cfg.get( "seed_items_per_table" , 500 ))
303
304 # Item-collection cardinality. items_per_partition > 1 means each
305 # partition key holds a real collection of that many items (distinct
306 # sort keys under a shared PK) instead of a singleton — so Query
307 # patterns read realistic multi-item pages and GSI collections are
308 # observable. Default 1 preserves the historical singleton behavior of
309 # quick/standard modes. The number of DISTINCT partitions is therefore
310 # seed_items // items_per_partition, and that partition count — not
311 # seed_items — is the space the read-key sampler draws over (a Query
312 # must target a partition that actually has a full collection seeded).
313 self .items_per_partition = max ( 1 , int ( self .cfg.get( "items_per_partition" , 1 )))
314 self .n_partitions = max ( 1 , self .seed_items // self .items_per_partition)
315
316 # Read-key sampling distribution. "uniform" (default) spreads load
317 # evenly via round-robin; "zipf" concentrates load on a few hot
318 # partitions so a single partition approaches the per-partition
319 # throughput ceiling (Mechanics #3, ~1000 WCU / 3000 RCU) and
320 # hot-partition throttling becomes observable.
321 self .key_sampling = ( self .cfg.get( "read_pattern_key_sampling" ) or "uniform" ).lower()
322 self .zipf_s = float ( self .cfg.get( "zipf_s" , 1.1 ))
323 self ._zipf_cum = None # lazily built cumulative-weight table
324 self ._zipf_lock = threading.Lock()
325
326 # Measurement window sizing.
327 self .duration_seconds = int ( self .cfg.get( "duration_seconds" , 90 ))
328 self .warmup_seconds = int ( self .cfg.get( "warmup_seconds" , 10 ))
329 self .ramp_seconds = int ( self .cfg.get( "ramp_seconds" , 10 ))
330 self .scale_factor = float ( self .cfg.get( "scale_factor" , 0.01 ))
331 self .min_rps = float ( self .cfg.get( "min_rps_per_pattern" , 1 ))
332 self .max_rps = float ( self .cfg.get( "max_rps_per_pattern" , 50 ))
333 # 32 worker threads per pattern. The per-pattern driver is an open-loop
334 # scheduler feeding a ThreadPoolExecutor; since each call is I/O-bound (a
335 # DynamoDB round trip), threads — not CPU — set the sustainable rate. At
336 # ~5-20ms/call, 32 threads sustain ~1500-2000 rps/pattern, raising the
337 # single-Lambda ceiling well above the old ~800-1000 (8 threads) before
338 # any multi-Lambda sharding would be needed. Overridable via config.
339 self .concurrency = int ( self .cfg.get( "concurrency_per_pattern" , 32 ))
340 self .abort_throttle = float ( self .cfg.get( "abort_on_throttle_rate" , 0.2 ))
341 # Non-throttle error rate that taints a pattern and stops its window. Set
342 # well above the throttle threshold — a structural error (bad index/attr,
343 # duplicate key, access denied) reliably fails ~100% of calls, so 0.5
344 # catches a genuinely broken pattern without tripping on sporadic
345 # transient errors. Overridable via config for tuning.
346 self .abort_error_rate = float ( self .cfg.get( "abort_on_error_rate" , 0.5 ))
347 self .table_settle = int ( self .cfg.get( "table_settle_seconds" , 30 ))
348
349 def prefixed_table (self, orig: str ) -> str :
350 """Return the benchmark-prefixed physical table name."""
351 return self .prefixed.get(orig, orig)
352
353 def vector_indexes (self, table_name: str ) -> list :
354 """Vector index metadata for a table, normalised for the seeder and dispatch.
355
356 Flattens search_schema so the driver does not have to know the design's nesting:
357 each entry carries vector_attribute, dimensions, partition_key, pk_type and
358 inline_filters.
359 """
360 spec = self .table_by_name.get(table_name) or {}
361 out = []
362 for vi in spec.get( "vector_indexes" ) or []:
363 schema = vi.get( "search_schema" ) or {}
364 pk = schema.get( "partition_key" )
365 out.append(
366 {
367 "index_name" : vi.get( "index_name" ),
368 "vector_attribute" : vi.get( "vector_attribute" ),
369 "dimensions" : vi.get( "dimensions" ),
370 "partition_key" : pk,
371 "pk_type" : self .key_type(table_name, pk) if pk else "S" ,
372 "inline_filters" : list (schema.get( "inline_filters" ) or []),
373 }
374 )
375 return out
376
377 def key_type (self, table_name: str , attr: str ) -> str :
378 """Declared DDB type ('S'/'N'/'B') of a key attribute; 'S' if unknown."""
379 return self .attr_types.get((table_name, attr), "S" )
380
381 def bench_rps_for (self, pattern) -> float :
382 declared = float (pattern.get( "peak_rps" , 0 ) or 0 )
383 scaled = declared * self .scale_factor
384 return max ( self .min_rps, min ( self .max_rps, scaled))
385
386 def _build_zipf_cum (self) -> list :
387 """Build a cumulative-probability table for ranks 1..n_partitions with
388 P(rank r) proportional to 1/r^s. Built once, under a lock, because the
389 submission loop calls sampled_idx() once per request at bench RPS."""
390 n = self .n_partitions
391 weights = [ 1.0 / (r ** self .zipf_s) for r in range ( 1 , n + 1 )]
392 total = sum (weights)
393 cum = []
394 acc = 0.0
395 for w in weights:
396 acc += w / total
397 cum.append(acc)
398 cum[ - 1 ] = 1.0 # guard against float drift so bisect always lands in-range
399 return cum
400
401 def sampled_idx (self, key_counter: int ) -> int :
402 """Map a per-call counter to a seeded PARTITION index in [0, n_partitions).
403
404 "uniform" → round-robin (historical behavior). "zipf" → draw a rank by
405 the precomputed cumulative distribution so a few partitions absorb most
406 traffic. The space is the number of distinct seeded partitions, so a
407 sampled index always points at a partition that has its full collection
408 seeded — never an unseeded key that would misread as empty/throttled."""
409 n = self .n_partitions
410 if n <= 1 :
411 return 0
412 if self .key_sampling == "zipf" :
413 if self ._zipf_cum is None :
414 with self ._zipf_lock:
415 if self ._zipf_cum is None :
416 self ._zipf_cum = self ._build_zipf_cum()
417 u = random.random()
418 return bisect.bisect_left( self ._zipf_cum, u)
419 return key_counter % n
420
421
422 # ---------------------------------------------------------------------------
423 # Raw-row recorder (thread-safe)
424 # ---------------------------------------------------------------------------
425
426
427 class RowSink :
428 def __init__ (self, n_patterns: int = 1 ):
429 self ._rows: list = []
430 self ._lock = threading.Lock()
431 self ._n_patterns = max ( 1 , int (n_patterns))
432 # Per-(pattern_id, phase) recorded counts — fair-share sub-cap within a
433 # phase so one busy pattern can't take the whole phase budget.
434 self ._per_key: dict = {}
435 # Per-phase recorded counts — the primary budget so warmup can't consume
436 # measure's allocation.
437 self ._phase_count: dict = {}
438 # Per-(pattern, phase) stack of row indices that are currently
439 # non-distressed (no throttle/error). Lets the distress-swap below run in
440 # O(1) instead of scanning every recorded row.
441 self ._noncrit: dict = {}
442
443 def _per_key_cap (self, phase: str ) -> int :
444 budget = PHASE_ROW_BUDGET .get(phase, 1_000 )
445 return max ( MIN_ROWS_PER_PATTERN_PHASE , budget // self ._n_patterns)
446
447 def add (self, row):
448 pid = row.get( "pattern_id" )
449 phase = row.get( "phase" )
450 key = (pid, phase)
451 with self ._lock:
452 n = self ._per_key.get(key, 0 )
453 phase_n = self ._phase_count.get(phase, 0 )
454 phase_budget = PHASE_ROW_BUDGET .get(phase, 1_000 )
455 # Append only if ALL three bounds allow it: the hard payload bound,
456 # this phase's reserved budget, and this pattern's fair share of it.
457 if (
458 len ( self ._rows) < MAX_ROWS_PER_INVOCATION
459 and phase_n < phase_budget
460 and n < self ._per_key_cap(phase)
461 ):
462 idx = len ( self ._rows)
463 self ._rows.append(row)
464 self ._per_key[key] = n + 1
465 self ._phase_count[phase] = phase_n + 1
466 if not (row.get( "throttled" ) or row.get( "error" )):
467 self ._noncrit.setdefault(key, []).append(idx)
468 return
469 # No room. A throttled/error row is diagnostically more valuable than
470 # yet another success — so swap it in over an earlier NON-distressed
471 # row for the same (pattern, phase). Otherwise the first N pre-throttle
472 # successes monopolize the sample and p99 looks clean even when the
473 # table is throttling hard. Exact call/throttle COUNTS are tracked
474 # separately and are unaffected. O(1) via the per-key index stack.
475 if not (row.get( "throttled" ) or row.get( "error" )):
476 return
477 stack = self ._noncrit.get(key)
478 while stack:
479 i = stack.pop()
480 existing = self ._rows[i]
481 if not (existing.get( "throttled" ) or existing.get( "error" )):
482 self ._rows[i] = row
483 return
484 # No swappable success row found — drop (sample already all-distress).
485
486 def extend (self, rows):
487 for r in rows:
488 self .add(r)
489
490 def drain (self):
491 with self ._lock:
492 out, self ._rows = self ._rows, []
493 self ._per_key = {}
494 self ._phase_count = {}
495 self ._noncrit = {}
496 return out
497
498
499 # ---------------------------------------------------------------------------
500 # Op dispatchers — each returns (consumed_cu_base, gsi_cu_by_index, latency_ms,
501 # throttled_bool, error_str)
502 # ---------------------------------------------------------------------------
503
504
505 def _consumed (cc_block):
506 """Extract (base CU, per-GSI CU dict) from ConsumedCapacity response.
507
508 DynamoDB's ConsumedCapacity shape:
509 - CapacityUnits (top-level) = total across base + LSI + GSI
510 - Table.CapacityUnits = base table only
511 - GlobalSecondaryIndexes.{name}.CapacityUnits = per-GSI
512 We want base-only so amplification_ratio = sum(GSI) / base is meaningful.
513 Prefer Table.CapacityUnits; fall back to top-level minus sum(GSI) if absent.
514
515 The Table and GlobalSecondaryIndexes blocks are only returned when the call asked
516 for ReturnConsumedCapacity='INDEXES'. Measured 2026-08-19: a PutItem fanning out to
517 one ALL-projection GSI reported CapacityUnits=6.0 with no sub-blocks under 'TOTAL',
518 versus Table=3.0 + GSI=3.0 under 'INDEXES'. So under 'TOTAL' the fallback below is
519 always taken, per-GSI comes back {}, and base absorbs the whole 6.0 — a 2x
520 overstatement, with amplification_ratio pinned at 0. That is why the call sites use
521 'INDEXES'; the fallback is now a genuine edge case rather than the normal path.
522 """
523 if not cc_block:
524 return 0.0 , {}
525 gsi = {}
526 for name, block in (cc_block.get( "GlobalSecondaryIndexes" ) or {}).items():
527 gsi[name] = float (block.get( "CapacityUnits" , 0.0 ) or 0.0 )
528 table_block = cc_block.get( "Table" ) or {}
529 if "CapacityUnits" in table_block:
530 base = float (table_block.get( "CapacityUnits" , 0.0 ) or 0.0 )
531 else :
532 top = float (cc_block.get( "CapacityUnits" , 0.0 ) or 0.0 )
533 base = max ( 0.0 , top - sum (gsi.values()))
534 return base, gsi
535
536
537 def _consumed_vector (cc_block):
538 """Extract vector byte counters from ConsumedCapacity.
539
540 Vector capacity is metered in BYTES, not capacity units, and appears in two places:
541 - SearchVectors: top-level "VectorSearchRequestBytes"
542 - writes: "VectorIndexes".{name}.VectorWriteRequestBytes
543
544 These are what make vector cost observable at all: search metering depends on index
545 traversal, so it cannot be derived from a design (see cost-model-schema.md). The
546 benchmark measuring it is the only sound source of a real number.
547
548 NOTE the per-index write values are each floored at 1 KB in the REPORTED payload,
549 independently of the request total, so summing them can overstate the billed amount
550 for sub-1KB vectors. Raw observations are recorded; interpretation lives in the report.
551 """
552 if not cc_block:
553 return 0.0 , {}
554 search_bytes = float (cc_block.get( "VectorSearchRequestBytes" , 0.0 ) or 0.0 )
555 per_index = {}
556 for name, block in (cc_block.get( "VectorIndexes" ) or {}).items():
557 per_index[name] = float ((block or {}).get( "VectorWriteRequestBytes" , 0.0 ) or 0.0 )
558 return search_bytes, per_index
559
560
561 def _time_call (fn, * args, ** kwargs):
562 start = time.monotonic()
563 try :
564 resp = fn( * args, ** kwargs)
565 lat_ms = (time.monotonic() - start) * 1000.0
566 cc_raw = resp.get( "ConsumedCapacity" ) if isinstance (resp, dict ) else None
567 # BatchWriteItem / BatchGetItem return ConsumedCapacity as a list of
568 # per-table dicts; single-item ops return a single dict. Normalise to
569 # one dict for _consumed.
570 if isinstance (cc_raw, list ):
571 merged: dict = {}
572 for entry in cc_raw:
573 for k, v in entry.items():
574 if k == "CapacityUnits" :
575 merged[ "CapacityUnits" ] = merged.get( "CapacityUnits" , 0.0 ) + float (v or 0 )
576 elif k == "Table" :
577 t = merged.setdefault( "Table" , {})
578 t[ "CapacityUnits" ] = t.get( "CapacityUnits" , 0.0 ) + float (
579 (v or {}).get( "CapacityUnits" , 0 ) or 0
580 )
581 elif k == "GlobalSecondaryIndexes" :
582 g = merged.setdefault( "GlobalSecondaryIndexes" , {})
583 for idx_name, idx_block in (v or {}).items():
584 g.setdefault(idx_name, {})
585 g[idx_name][ "CapacityUnits" ] = g[idx_name].get(
586 "CapacityUnits" , 0.0
587 ) + float ((idx_block or {}).get( "CapacityUnits" , 0 ) or 0 )
588 elif k == "VectorIndexes" :
589 vx = merged.setdefault( "VectorIndexes" , {})
590 for idx_name, idx_block in (v or {}).items():
591 vx.setdefault(idx_name, {})
592 vx[idx_name][ "VectorWriteRequestBytes" ] = vx[idx_name].get(
593 "VectorWriteRequestBytes" , 0.0
594 ) + float ((idx_block or {}).get( "VectorWriteRequestBytes" , 0 ) or 0 )
595 elif k == "VectorSearchRequestBytes" :
596 merged[ "VectorSearchRequestBytes" ] = merged.get(
597 "VectorSearchRequestBytes" , 0.0
598 ) + float (v or 0 )
599 cc_block = merged if merged else None
600 else :
601 cc_block = cc_raw
602 base, gsi = _consumed(cc_block)
603 vec_search_bytes, vec_write_bytes = _consumed_vector(cc_block)
604 return {
605 "latency_ms" : lat_ms,
606 "consumed_cu" : base,
607 "gsi_cu" : gsi,
608 "vector_search_bytes" : vec_search_bytes,
609 "vector_write_bytes" : vec_write_bytes,
610 "throttled" : False ,
611 "error" : None ,
612 "resp" : resp,
613 }
614 except ClientError as e:
615 lat_ms = (time.monotonic() - start) * 1000.0
616 code = e.response.get( "Error" , {}).get( "Code" , "" )
617 # TransactWriteItems/TransactGetItems surface a single top-level
618 # "TransactionCanceledException" whose ACTUAL per-item reasons live in
619 # e.response["CancellationReasons"] (a list of {Code, Message}). The
620 # top-level code alone can't tell a TransactionConflict (concurrent
621 # writes to the same key — a benchmark artifact on a small seeded key
622 # space) from a ConditionalCheckFailed (a guard the design relies on)
623 # from a real ValidationException. Extract the distinct, non-"None"
624 # reason codes so the report narrates from data instead of guessing.
625 cancel_codes = [
626 r.get( "Code" )
627 for r in (e.response.get( "CancellationReasons" ) or [])
628 if r.get( "Code" ) and r.get( "Code" ) != "None"
629 ]
630 return {
631 "latency_ms" : lat_ms,
632 "consumed_cu" : 0.0 ,
633 "gsi_cu" : {},
634 "vector_search_bytes" : 0.0 ,
635 "vector_write_bytes" : {},
636 "throttled" : code in THROTTLE_CODES ,
637 "error" : code,
638 "cancellation_reasons" : cancel_codes,
639 "resp" : None ,
640 }
641 except Exception as e:
642 lat_ms = (time.monotonic() - start) * 1000.0
643 return {
644 "latency_ms" : lat_ms,
645 "consumed_cu" : 0.0 ,
646 "gsi_cu" : {},
647 "vector_search_bytes" : 0.0 ,
648 "vector_write_bytes" : {},
649 "throttled" : False ,
650 "error" : type (e). __name__ ,
651 "resp" : None ,
652 }
653
654
655 def _dispatch (ctx: RunContext, pattern: dict , part_idx: int , member_counter: int ):
656 """Dispatch one call of the declared operation. Never substitute op types.
657
658 `part_idx` is the seeded PARTITION index chosen by the caller via
659 ctx.sampled_idx() (uniform round-robin or zipf hot-key skew) — passed in so
660 the caller can record it for the key-distribution histogram. `member_counter`
661 selects which collection member within the partition a point op targets.
662 The result dict carries `sampled_partition` so the drain can build the
663 per-pattern histogram that surfaces hot-partition skew (Mechanics #3)."""
664 op = pattern[ "operation" ]
665 td = ctx.table_by_name.get(pattern[ "table" ])
666 if not td:
667 return {
668 "latency_ms" : 0.0 ,
669 "consumed_cu" : 0.0 ,
670 "gsi_cu" : {},
671 "throttled" : False ,
672 "error" : "table_not_found_in_design" ,
673 }
674 ks = td.get( "key_schema" ) or {}
675 pk_attr = ks.get( "partition_key" )
676 sk_attr = ks.get( "sort_key" )
677 table_name = ctx.prefixed_table(td[ "table_name" ])
678 pid = pattern[ "pattern_id" ]
679 item_size = int (pattern.get( "estimated_item_size_bytes" , 1024 ))
680 items_per = int (pattern.get( "items_per_request" , 1 ))
681
682 # Key layout mirrors run_seed: partition `pidx` holds `items_per_partition`
683 # members; the global sort-key index is pidx*ipp + member. A single point op
684 # targets (part_idx, member_counter % ipp); multi-key ops (batch/transact)
685 # walk distinct (partition, member) pairs starting at part_idx so every key
686 # they touch is one that was actually seeded.
687 ipp = ctx.items_per_partition
688 npar = ctx.n_partitions
689
690 # Resolve key attribute types once so every generated key value matches the
691 # declared type (a numeric key rejects a string value — the bug a "recorded_at"
692 # N sort key hit). Defaults to "S".
693 pk_type = ctx.key_type(td[ "table_name" ], pk_attr or "" )
694 sk_type = ctx.key_type(td[ "table_name" ], sk_attr) if sk_attr else "S"
695
696 def _pk (idx):
697 return _seed_key_val(pid, idx, "pk" , pk_type)
698
699 def _sk (idx):
700 return _seed_key_val(pid, idx, "sk" , sk_type)
701
702 def _key_at (j):
703 pidx = (part_idx + j) % npar
704 member = ((member_counter + j) % ipp) if ipp > 1 else 0
705 global_idx = pidx * ipp + member
706 pkv = _pk(pidx)
707 skv = _sk(global_idx) if sk_attr else None
708 return pkv, skv
709
710 # Distinct seeded-key space, matching run_seed exactly:
711 # sort-key table -> npar partitions × ipp members = npar*ipp distinct
712 # (pk, sk) pairs; flat index g maps pidx=g//ipp,
713 # global_idx=g.
714 # no-sort-key table-> run_seed writes one item per partition over
715 # ctx.seed_items partitions, so there are seed_items
716 # distinct partition keys (member is meaningless with
717 # no SK).
718 # A flat index walked modulo this count yields ONLY distinct keys, so
719 # multi-key requests can never contain a duplicate (DynamoDB rejects a whole
720 # BatchGetItem/BatchWriteItem/Transact* request that lists the same key
721 # twice — "Provided list of item keys contains duplicates"). This replaces
722 # the old (part_idx+j)%npar walk, which wrapped onto an already-used key
723 # whenever items_per_request exceeded the distinct space (the common case on
724 # a no-SK table where the space is just `npar` partitions).
725 n_distinct_keys = (npar * ipp) if sk_attr else ctx.seed_items
726
727 def _key_flat (g):
728 gg = g % max ( 1 , n_distinct_keys)
729 if sk_attr:
730 pidx = gg // ipp
731 return _pk(pidx), _sk(gg)
732 return _pk(gg), None
733
734 def _distinct_keys (n_requested, hard_max):
735 """Yield (pkv, skv) for `n` DISTINCT seeded keys, starting at the hot
736 partition so zipf skew is preserved. `n` is capped at the distinct
737 seeded-key space (so we never duplicate) and at the op's DynamoDB limit
738 (`hard_max`). Returns a list; len may be < n_requested when the seeded
739 space or the API limit is smaller."""
740 n = max ( 1 , min ( int (n_requested), int (hard_max), int (n_distinct_keys)))
741 if sk_attr:
742 start_g = (part_idx % max ( 1 , npar)) * ipp + (member_counter % ipp)
743 else :
744 start_g = part_idx % max ( 1 , n_distinct_keys)
745 return [_key_flat(start_g + k) for k in range (n)]
746
747 pk_val, sk_val = _key_at( 0 )
748 idx = part_idx # partition index for GSI-synthetic key derivation
749
750 client = ctx.client
751
752 if op == "GetItem" :
753 key = {pk_attr: _serialize(pk_val)}
754 if sk_attr:
755 key[sk_attr] = _serialize(sk_val)
756 consistent = pattern.get( "consistency" ) == "strong"
757 return _time_call(
758 client.get_item,
759 TableName = table_name,
760 Key = key,
761 ConsistentRead = consistent,
762 ReturnConsumedCapacity = "INDEXES" ,
763 )
764
765 if op == "Query" :
766 # Query on base table OR on a GSI.
767 index = pattern.get( "index" )
768 if index:
769 # Query GSI: we seeded base items with GSI PK populated. Use
770 # the GSI's partition key attribute.
771 gsi_def = next ((g for g in td.get( "gsis" ) or [] if g[ "index_name" ] == index), None )
772 if not gsi_def:
773 return {
774 "latency_ms" : 0.0 ,
775 "consumed_cu" : 0.0 ,
776 "gsi_cu" : {},
777 "throttled" : False ,
778 "error" : "gsi_not_found" ,
779 }
780 g_pk = gsi_def[ "partition_key" ]
781 # If the GSI shares its PK attribute with the base table's PK or
782 # SK, seeding did NOT set a synthetic gsi value (see run_seed) —
783 # so query with the base PK/SK value. Otherwise use the synthetic.
784 if g_pk == pk_attr:
785 g_pk_val = pk_val
786 elif g_pk == sk_attr:
787 g_pk_val = sk_val
788 else :
789 # Synthetic GSI PK — seeded off the partition index (pidx==idx),
790 # type-aware, IDENTICAL to run_seed's _gsi_val so the Query hits.
791 g_pk_val = _gsi_val(pid, idx, "pk" , ctx.key_type(td[ "table_name" ], g_pk))
792 kwargs = dict (
793 TableName = table_name,
794 IndexName = index,
795 KeyConditionExpression = "#pk = :pk" ,
796 ExpressionAttributeNames = { "#pk" : g_pk},
797 ExpressionAttributeValues = { ":pk" : _serialize(g_pk_val)},
798 Limit = items_per,
799 ReturnConsumedCapacity = "INDEXES" ,
800 )
801 else :
802 kwargs = dict (
803 TableName = table_name,
804 KeyConditionExpression = "#pk = :pk" ,
805 ExpressionAttributeNames = { "#pk" : pk_attr},
806 ExpressionAttributeValues = { ":pk" : _serialize(pk_val)},
807 Limit = items_per,
808 ReturnConsumedCapacity = "INDEXES" ,
809 )
810 return _time_call(client.query, ** kwargs)
811
812 if op == "Scan" :
813 # One-shot per Mechanics #16 — do not loop. Limit items to the
814 # declared items_per_request to keep costs predictable.
815 return _time_call(
816 client.scan,
817 TableName = table_name,
818 Limit = max ( 1 , items_per),
819 ReturnConsumedCapacity = "INDEXES" ,
820 )
821
822 if op == "BatchGetItem" :
823 # BatchGetItem caps at 100 keys per call; keys must be distinct.
824 keys = []
825 for bpk, bsk in _distinct_keys(items_per, hard_max = 100 ):
826 k = {pk_attr: _serialize(bpk)}
827 if sk_attr:
828 k[sk_attr] = _serialize(bsk)
829 keys.append(k)
830 return _time_call(
831 client.batch_get_item,
832 RequestItems = {table_name: { "Keys" : keys}},
833 ReturnConsumedCapacity = "INDEXES" ,
834 )
835
836 if op == "PutItem" :
837 # Annotated because the vector branch below adds an embedding list and a
838 # partition-key string; inferred from this literal alone it would be dict[str, int].
839 write_extra: dict[ str , Any] = { "bench_ts" : int (time.time())}
840 # Without these the write never reaches the vector index, so the run reports no
841 # vector write capacity and the index looks free to maintain. The partition key
842 # matters as much as the vector: omit it and the write SUCCEEDS while the item is
843 # silently excluded from the index.
844 for vi in ctx.vector_indexes(pattern[ "table" ]):
845 vattr, vdims = vi.get( "vector_attribute" ), int (vi.get( "dimensions" ) or 0 )
846 if vattr and vdims:
847 vrng = random.Random( f " { pattern[ 'pattern_id' ] } : { part_idx } : { vattr } :w" )
848 write_extra[vattr] = [ round (vrng.uniform( - 1.0 , 1.0 ), 6 ) for _ in range (vdims)]
849 vpk = vi.get( "partition_key" )
850 if vpk and vpk not in (pk_attr, sk_attr):
851 write_extra[vpk] = _seed_key_val(
852 pattern[ "pattern_id" ], part_idx, "vector_pk" , vi.get( "pk_type" , "S" )
853 )
854 item = _build_item(
855 pk_attr,
856 pk_val,
857 sk_attr,
858 sk_val,
859 item_size,
860 extra = write_extra,
861 )
862 return _time_call(
863 client.put_item,
864 TableName = table_name,
865 Item = item,
866 ReturnConsumedCapacity = "INDEXES" ,
867 )
868
869 if op == "UpdateItem" :
870 # Update a single attribute — mirrors a typical mutate-one-field call.
871 key = {pk_attr: _serialize(pk_val)}
872 if sk_attr:
873 key[sk_attr] = _serialize(sk_val)
874 return _time_call(
875 client.update_item,
876 TableName = table_name,
877 Key = key,
878 UpdateExpression = "SET bench_ts = :t" ,
879 ExpressionAttributeValues = { ":t" : _serialize( int (time.time()))},
880 ReturnConsumedCapacity = "INDEXES" ,
881 )
882
883 if op == "DeleteItem" :
884 key = {pk_attr: _serialize(pk_val)}
885 if sk_attr:
886 key[sk_attr] = _serialize(sk_val)
887 return _time_call(
888 client.delete_item,
889 TableName = table_name,
890 Key = key,
891 ReturnConsumedCapacity = "INDEXES" ,
892 )
893
894 if op == "BatchWriteItem" :
895 # BatchWriteItem caps at 25 items per call; PutRequests in one call must
896 # not target duplicate primary keys.
897 reqs = []
898 for bpk, bsk in _distinct_keys(items_per, hard_max = 25 ):
899 it = _build_item(
900 pk_attr,
901 bpk,
902 sk_attr,
903 bsk,
904 item_size,
905 extra = { "bench_ts" : int (time.time())},
906 )
907 reqs.append({ "PutRequest" : { "Item" : it}})
908 return _time_call(
909 client.batch_write_item,
910 RequestItems = {table_name: reqs},
911 ReturnConsumedCapacity = "INDEXES" ,
912 )
913
914 if op == "TransactWriteItems" :
915 # Use declared item_sizes if present; fall back to items_per × item_size.
916 # A transaction caps at 100 items and cannot operate on the SAME item
917 # twice ("Transaction request cannot include multiple operations on one
918 # item"), so walk DISTINCT seeded keys. The number of writes is the count
919 # of declared sizes, still capped at the distinct seeded space.
920 sizes = pattern.get( "item_sizes" ) or [item_size] * max ( 1 , items_per)
921 keys = _distinct_keys( len (sizes), hard_max = 100 )
922 tx_items = []
923 for i, (bpk, bsk) in enumerate (keys):
924 sz = sizes[i] if i < len (sizes) else item_size
925 tx_items.append(
926 {
927 "Put" : {
928 "TableName" : table_name,
929 "Item" : _build_item(
930 pk_attr,
931 bpk,
932 sk_attr,
933 bsk,
934 sz,
935 extra = { "bench_tx" : int (time.time()), "i" : i},
936 ),
937 }
938 }
939 )
940 return _time_call(
941 client.transact_write_items,
942 TransactItems = tx_items,
943 ReturnConsumedCapacity = "INDEXES" ,
944 )
945
946 if op == "TransactGetItems" :
947 # TransactGetItems caps at 100 items; keys must be distinct.
948 tx_items = []
949 for bpk, bsk in _distinct_keys(items_per, hard_max = 100 ):
950 k = {pk_attr: _serialize(bpk)}
951 if sk_attr:
952 k[sk_attr] = _serialize(bsk)
953 tx_items.append({ "Get" : { "TableName" : table_name, "Key" : k}})
954 return _time_call(
955 client.transact_get_items,
956 TransactItems = tx_items,
957 ReturnConsumedCapacity = "INDEXES" ,
958 )
959
960 if op == VECTOR_SEARCH_OP :
961 # The pattern carries the index metadata the driver needs, injected by
962 # benchmark_model.py from the design: dimensions (the query vector must match
963 # exactly, or every call fails), the SearchSchema partition key (mandatory in
964 # SearchConditionExpression when the index defines one), and TopK.
965 vi = pattern.get( "vector_index" ) or {}
966 dims = int (vi.get( "dimensions" ) or 0 )
967 if not hasattr (client, "search_vectors" ):
968 # The runtime's boto3 predates vector support. Left to itself this raises a
969 # bare AttributeError that the drain loop records as a generic exception, so
970 # the pattern shows 0 consumed capacity and reads as FREE. Return an explicit,
971 # named error instead so the aggregation and report can call it out.
972 import botocore as _bc
973
974 return {
975 "latency_ms" : 0.0 ,
976 "consumed_cu" : 0.0 ,
977 "gsi_cu" : {},
978 "vector_search_bytes" : 0.0 ,
979 "vector_write_bytes" : {},
980 "throttled" : False ,
981 "error" : f "search_vectors_unavailable_botocore_ { _bc. __version__ } " ,
982 }
983 if not dims:
984 return {
985 "latency_ms" : 0.0 ,
986 "consumed_cu" : 0.0 ,
987 "gsi_cu" : {},
988 "vector_search_bytes" : 0.0 ,
989 "vector_write_bytes" : {},
990 "throttled" : False ,
991 "error" : "vector_index_metadata_missing" ,
992 }
993 # A deterministic pseudo-random unit-ish vector. Content does not affect
994 # metering — measurement showed bytes depend on TopK, projection and dimensions,
995 # not on the query vector's values — so a fixed seed keeps runs comparable.
996 rng = random.Random( f " { pattern[ 'pattern_id' ] } : { part_idx } " )
997 query_vector = [{ "N" : f " { rng.uniform( - 1.0 , 1.0 ) :.6f} " } for _ in range (dims)]
998
999 kwargs = {
1000 "TableName" : table_name,
1001 "IndexName" : pattern.get( "index" ),
1002 "SearchVector" : query_vector,
1003 "TopK" : max ( 1 , min ( int (pattern.get( "top_k" ) or 10 ), 100 )),
1004 "ReturnConsumedCapacity" : "INDEXES" ,
1005 }
1006 v_pk = vi.get( "partition_key" )
1007 if v_pk:
1008 # Scoped to the same seeded partition the read sampler chose, so the search
1009 # examines seeded data rather than an empty partition. Aliased via
1010 # ExpressionAttributeNames because SearchConditionExpression enforces
1011 # reserved words exactly as other expression parameters do.
1012 #
1013 # The coincidence cases are load-bearing, exactly as they are for a GSI Query
1014 # above. When the search-schema partition key IS the table's PK or SK, neither
1015 # run_seed nor the PutItem path writes a separate synthetic "vector_pk" value —
1016 # the attribute already holds the base key value — so searching for the
1017 # synthetic one queries a partition nothing was ever written to. The search
1018 # returns no results, VectorSearchRequestBytes reports ~0, and the index reads
1019 # as FREE, which is the precise failure this instrumentation exists to prevent
1020 # and is indistinguishable from a genuinely cheap index.
1021 if v_pk == pk_attr:
1022 v_pk_val = pk_val
1023 elif v_pk == sk_attr:
1024 v_pk_val = sk_val
1025 else :
1026 v_pk_val = _seed_key_val(
1027 pattern[ "pattern_id" ], part_idx, "vector_pk" , vi.get( "pk_type" , "S" )
1028 )
1029 kwargs[ "SearchConditionExpression" ] = "#vpk = :vpk"
1030 kwargs[ "ExpressionAttributeNames" ] = { "#vpk" : v_pk}
1031 kwargs[ "ExpressionAttributeValues" ] = { ":vpk" : _serialize(v_pk_val)}
1032 return _time_call(client.search_vectors, ** kwargs)
1033
1034 return {
1035 "latency_ms" : 0.0 ,
1036 "consumed_cu" : 0.0 ,
1037 "gsi_cu" : {},
1038 "vector_search_bytes" : 0.0 ,
1039 "vector_write_bytes" : {},
1040 "throttled" : False ,
1041 "error" : f "unsupported_op: { op } " ,
1042 }
1043
1044
1045 # ---------------------------------------------------------------------------
1046 # Phase runners
1047 # ---------------------------------------------------------------------------
1048
1049
1050 def run_settle (ctx: RunContext, sink: RowSink):
1051 """Wait for cold-start capacity to stabilize; warm each thread's SDK."""
1052 if ctx.table_settle > 0 :
1053 time.sleep(ctx.table_settle)
1054
1055 def _warm (table_name):
1056 start = time.monotonic()
1057 try :
1058 ctx.client.describe_table( TableName = table_name)
1059 except ClientError:
1060 pass
1061 return (time.monotonic() - start) * 1000.0
1062
1063 prefixed_names = list (ctx.prefixed.values())
1064 with ThreadPoolExecutor( max_workers = min ( 8 , max ( 1 , len (prefixed_names)))) as pool:
1065 for tn in prefixed_names:
1066 lat = pool.submit(_warm, tn).result()
1067 sink.add(
1068 {
1069 "pattern_id" : "__settle__" ,
1070 "op" : "DescribeTable" ,
1071 "phase" : "settle" ,
1072 "ts" : time.time(),
1073 "latency_ms" : lat,
1074 "consumed_cu" : 0.0 ,
1075 "gsi_cu" : {},
1076 "throttled" : False ,
1077 "error" : None ,
1078 "table" : tn,
1079 }
1080 )
1081
1082
1083 def _tables_in_use (ctx: RunContext) -> set[ str ]:
1084 """Every table referenced by at least one access pattern (read OR write)."""
1085 return {p[ "table" ] for p in ctx.patterns if p.get( "table" )}
1086
1087
1088 def run_seed (ctx: RunContext, sink: RowSink) -> dict :
1089 """Seed every table that has any access pattern. Returns per-table seed counts."""
1090 seed_count_per_table = {}
1091 patterns_by_table: dict[ str , list[ dict ]] = {}
1092 for p in ctx.patterns:
1093 patterns_by_table.setdefault(p[ "table" ], []).append(p)
1094
1095 for orig_table in _tables_in_use(ctx):
1096 td = ctx.table_by_name.get(orig_table)
1097 if not td:
1098 continue
1099 ks = td.get( "key_schema" ) or {}
1100 pk_attr = ks.get( "partition_key" )
1101 sk_attr = ks.get( "sort_key" )
1102 table_name = ctx.prefixed_table(orig_table)
1103 table_patterns = patterns_by_table.get(orig_table, [])
1104
1105 # Determine an item size for seeds: use the largest item size declared
1106 # for any pattern on this table so reads against seeded items look
1107 # realistic.
1108 item_size = max (
1109 ( int (p.get( "estimated_item_size_bytes" , 1024 )) for p in table_patterns),
1110 default = 1024 ,
1111 )
1112
1113 # For GSI queries on this table, also populate the GSI PK/SK with a
1114 # predictable pattern-scoped value so the Query phase hits items.
1115 gsis = {g[ "index_name" ]: g for g in (td.get( "gsis" ) or [])}
1116
1117 # Seed items per pattern that reads from this table, so every read
1118 # pattern has items to hit. Write-only patterns still get seeded
1119 # (UpdateItem/DeleteItem target pre-existing items).
1120 #
1121 # Layout: n_partitions distinct partition keys, each holding
1122 # items_per_partition members (distinct sort keys). The global sort-key
1123 # index is pidx*ipp + member — the SAME mapping _dispatch._key_at uses,
1124 # so a Query on partition `pidx` reads the full seeded collection and a
1125 # point op resolves to a real member. With items_per_partition=1 this
1126 # reduces to the historical one-item-per-partition behavior.
1127 #
1128 # A table with NO sort key cannot hold a collection (no second key to
1129 # vary), so it falls back to one item per partition regardless of ipp.
1130 ipp = ctx.items_per_partition if sk_attr else 1
1131 n_part = ctx.n_partitions if sk_attr else ctx.seed_items
1132 # Resolve key types so seeded primary keys match _dispatch's generated
1133 # keys EXACTLY (same type, same value) — otherwise a Query/GetItem would
1134 # look for a key the seed never wrote. Defaults to "S".
1135 pk_type = ctx.key_type(orig_table, pk_attr or "" )
1136 sk_type = ctx.key_type(orig_table, sk_attr) if sk_attr else "S"
1137 items = []
1138 for p in table_patterns:
1139 pid = p[ "pattern_id" ]
1140 idx_name = p.get( "index" )
1141 g = gsis.get(idx_name) if idx_name else None
1142 for pidx in range (n_part):
1143 for member in range (ipp):
1144 global_idx = pidx * ipp + member
1145 gsi_extras = {}
1146 if g:
1147 g_pk = g[ "partition_key" ]
1148 g_sk = g.get( "sort_key" )
1149 # Key the GSI PK off the PARTITION (pidx), not the member,
1150 # so the GSI also holds a real collection per partition —
1151 # this is what makes GSI Query and GSI amplification
1152 # observable at volume. Never overwrite an attribute that
1153 # is the base table's own PK/SK (that would corrupt the
1154 # primary key); only set distinct GSI attributes. GSI key
1155 # values are type-aware too (a numeric GSI key rejects a
1156 # string), matching the _dispatch Query side.
1157 if g_pk != pk_attr and g_pk != sk_attr:
1158 gsi_extras[g_pk] = _gsi_val(
1159 pid, pidx, "pk" , ctx.key_type(orig_table, g_pk)
1160 )
1161 if g_sk and g_sk != pk_attr and g_sk != sk_attr:
1162 gsi_extras[g_sk] = _gsi_val(
1163 pid, global_idx, "sk" , ctx.key_type(orig_table, g_sk)
1164 )
1165 # Vector attributes. Two omissions here would BOTH look like a
1166 # cheap, working benchmark rather than an error:
1167 # * no vector attribute -> the item is never replicated to the
1168 # index, so SearchVectors returns nothing and reads as free
1169 # * no SearchSchema partition key -> the write SUCCEEDS on the base
1170 # table with no error and the item is silently de-indexed
1171 # So both are set explicitly, and the vector length must match the
1172 # index's Dimensions exactly or the write is rejected outright.
1173 vec_extras = {}
1174 for vi in ctx.vector_indexes(orig_table):
1175 vattr = vi.get( "vector_attribute" )
1176 vdims = int (vi.get( "dimensions" ) or 0 )
1177 if vattr and vdims:
1178 vrng = random.Random( f " { pid } : { pidx } : { global_idx } : { vattr } " )
1179 vec_extras[vattr] = [
1180 round (vrng.uniform( - 1.0 , 1.0 ), 6 ) for _ in range (vdims)
1181 ]
1182 vpk = vi.get( "partition_key" )
1183 if vpk and vpk not in (pk_attr, sk_attr) and vpk not in gsi_extras:
1184 # Same value the SearchVectors dispatch will search for, so
1185 # the search hits a populated partition.
1186 vec_extras[vpk] = _seed_key_val(
1187 pid, pidx, "vector_pk" , vi.get( "pk_type" , "S" )
1188 )
1189 for f in vi.get( "inline_filters" ) or []:
1190 if f not in (pk_attr, sk_attr) and f not in gsi_extras:
1191 vec_extras.setdefault(f, _gsi_val(pid, pidx, "vfilter" , "S" ))
1192
1193 extra = { "bench_seed" : 1 }
1194 extra.update(vec_extras)
1195 it = _build_item(
1196 pk_attr,
1197 _seed_key_val(pid, pidx, "pk" , pk_type),
1198 sk_attr,
1199 _seed_key_val(pid, global_idx, "sk" , sk_type) if sk_attr else None ,
1200 item_size,
1201 gsi_attrs = gsi_extras,
1202 extra = extra,
1203 )
1204 items.append(it)
1205
1206 # Deduplicate by primary key as a safety net. Each pattern's keys embed
1207 # its own pattern_id (via _seed_key_val(pid, …)), so patterns sharing a
1208 # table do NOT collide — this table gets ~seed_items items PER pattern, by design
1209 # (each pattern reads its own seeded keyspace via _dispatch._key_at). The
1210 # dedup only guards against an accidental intra-pattern collision; it is
1211 # effectively a no-op for the current key layout. (The spend estimate in
1212 # benchmark_model._estimate_bench_spend accounts for the per-pattern
1213 # volume; do not assume cross-pattern dedup shrinks it.)
1214 def _keyval (av):
1215 # The single scalar value out of a DDB attribute-value dict,
1216 # whatever its type tag (S/N/B) — so dedup works for numeric keys
1217 # too, not just strings.
1218 if not av:
1219 return None
1220 return next ( iter (av.values()))
1221
1222 seen_keys = set ()
1223 deduped = []
1224 for it in items:
1225 key_tuple = (_keyval(it.get(pk_attr)), _keyval(it.get(sk_attr)) if sk_attr else None )
1226 if key_tuple in seen_keys:
1227 continue
1228 seen_keys.add(key_tuple)
1229 deduped.append(it)
1230
1231 # Write in BatchWriteItem chunks of 25.
1232 written = 0
1233 for chunk_start in range ( 0 , len (deduped), 25 ):
1234 chunk = deduped[chunk_start : chunk_start + 25 ]
1235 reqs = [{ "PutRequest" : { "Item" : it}} for it in chunk]
1236 res = _time_call(
1237 ctx.client.batch_write_item,
1238 RequestItems = {table_name: reqs},
1239 ReturnConsumedCapacity = "INDEXES" ,
1240 )
1241 sink.add(
1242 {
1243 "pattern_id" : "__seed__" ,
1244 "op" : "BatchWriteItem" ,
1245 "phase" : "seed" ,
1246 "ts" : time.time(),
1247 "latency_ms" : res[ "latency_ms" ],
1248 "consumed_cu" : res[ "consumed_cu" ],
1249 "gsi_cu" : res[ "gsi_cu" ],
1250 # Vector capacity is metered in bytes, not CU — carried through
1251 # because a SearchVectors pattern has no other cost signal.
1252 "vector_search_bytes" : res.get( "vector_search_bytes" , 0.0 ),
1253 "vector_write_bytes" : res.get( "vector_write_bytes" ) or {},
1254 "throttled" : res[ "throttled" ],
1255 "error" : res[ "error" ],
1256 "table" : table_name,
1257 }
1258 )
1259 # Handle UnprocessedItems with bounded exponential backoff. Seeding
1260 # is correctness, not measurement — and now that the data-path client
1261 # has retries disabled (so the MEASURE phase can observe throttles),
1262 # seeding must do its own retry, especially against a low PROVISIONED
1263 # capacity where BatchWriteItem will shed items until capacity frees.
1264 if res.get( "resp" ):
1265 unproc = res[ "resp" ].get( "UnprocessedItems" ) or {}
1266 attempt = 0
1267 while unproc.get(table_name) and attempt < 8 :
1268 time.sleep( min ( 0.1 * ( 2 ** attempt), 3.0 ))
1269 retry = _time_call(
1270 ctx.client.batch_write_item,
1271 RequestItems = unproc,
1272 ReturnConsumedCapacity = "INDEXES" ,
1273 )
1274 sink.add(
1275 {
1276 "pattern_id" : "__seed_retry__" ,
1277 "op" : "BatchWriteItem" ,
1278 "phase" : "seed" ,
1279 "ts" : time.time(),
1280 "latency_ms" : retry[ "latency_ms" ],
1281 "consumed_cu" : retry[ "consumed_cu" ],
1282 "gsi_cu" : retry[ "gsi_cu" ],
1283 "throttled" : retry[ "throttled" ],
1284 "error" : retry[ "error" ],
1285 "table" : table_name,
1286 }
1287 )
1288 unproc = (retry.get( "resp" ) or {}).get( "UnprocessedItems" ) or {}
1289 attempt += 1
1290 written += len (chunk)
1291
1292 seed_count_per_table[table_name] = written
1293
1294 return seed_count_per_table
1295
1296
1297 SEED_VERIFY_CAP = 2000 # max items counted per table — bounds the verify cost
1298
1299
1300 def verify_seed (ctx: RunContext, seed_counts: dict ) -> dict :
1301 """Verify seed landed via a bounded-pagination Scan(Select=COUNT).
1302
1303 DescribeTable.ItemCount is eventually consistent — updated ~every 6h — so a
1304 freshly-seeded table always reports 0. A Scan with Select=COUNT returns a
1305 ground-truth count of what landed. The OLD implementation used Limit=1, so
1306 `Count` capped at 1 and `passed = actual > 0` was true whenever a SINGLE
1307 item landed — a table that seeded 1 of 1000 items read as clean, hiding a
1308 massive shortfall that silently corrupts every measurement on that table.
1309
1310 Now we paginate, accumulating `Count`, until we either reach the target
1311 (`min(expected, SEED_VERIFY_CAP)`) or exhaust the table. Cost is bounded:
1312 Select=COUNT bills on items examined, capped at ~SEED_VERIFY_CAP items.
1313
1314 Per-table result fields:
1315 expected declared seed target
1316 actual items counted (≥ this many exist; '+' if we stopped at cap)
1317 sampled True if we stopped at the cap before exhausting the table
1318 passed expected==0, OR actual ≥ 50% of the capped target
1319 seed_shortfall_ratio 1 - actual/expected (0.0 when expected==0); only
1320 meaningful when not `sampled`
1321 """
1322 out = {}
1323 for table_name, expected in seed_counts.items():
1324 target = min ( int (expected), SEED_VERIFY_CAP ) if expected else 0
1325 actual = 0
1326 sampled = False
1327 start_key = None
1328 try :
1329 while True :
1330 kwargs = { "TableName" : table_name, "Select" : "COUNT" }
1331 if start_key:
1332 kwargs[ "ExclusiveStartKey" ] = start_key
1333 resp = ctx.client.scan( ** kwargs)
1334 actual += resp.get( "Count" , 0 )
1335 start_key = resp.get( "LastEvaluatedKey" )
1336 if target and actual >= target:
1337 # Counted enough to make the pass/fail call; stop early so a
1338 # large table doesn't run up cost past the cap.
1339 sampled = bool (start_key)
1340 break
1341 if not start_key:
1342 break # exhausted the table — `actual` is exact
1343 except ClientError:
1344 actual = 0
1345 # Threshold against the CAPPED target, not raw expected, so a table with
1346 # more than SEED_VERIFY_CAP declared items isn't failed for our choosing
1347 # to stop counting.
1348 passed = expected == 0 or actual >= max ( 1 , int (target * 0.5 ))
1349 shortfall = ( 1.0 - (actual / expected)) if expected else 0.0
1350 out[table_name] = {
1351 "expected" : expected,
1352 "actual" : actual,
1353 "sampled" : sampled,
1354 "passed" : passed,
1355 "seed_shortfall_ratio" : round ( max ( 0.0 , shortfall), 3 ),
1356 }
1357 return out
1358
1359
1360 def run_pattern_window (
1361 ctx: RunContext,
1362 pattern: dict ,
1363 duration_s: float ,
1364 phase: str ,
1365 sink: RowSink,
1366 tainted: dict ,
1367 counts: dict | None = None ,
1368 ):
1369 """Drive one pattern at its bench RPS for duration_s seconds.
1370
1371 Scheduler design: a single submission thread submits calls at the
1372 configured rate to a ThreadPoolExecutor. A separate drain thread pulls
1373 completed futures off a queue and records rows. This keeps the
1374 submission loop from blocking on result processing, so genuine
1375 concurrency matches `concurrency_per_pattern`.
1376
1377 `counts`, if given, receives EXACT (uncapped) call/throttle tallies keyed by
1378 (pattern_id, phase). Recorded rows are per-key capped for the response
1379 payload, but these counts see every call — so throttle totals stay accurate
1380 even when the latency-percentile rows are down-sampled.
1381 """
1382 import queue
1383
1384 pid = pattern[ "pattern_id" ]
1385 bench_rps = ctx.bench_rps_for(pattern)
1386 if bench_rps <= 0 :
1387 return
1388
1389 deadline = time.monotonic() + duration_s
1390 interval = 1.0 / bench_rps
1391 ramp_deadline = time.monotonic() + ctx.ramp_seconds if phase == "measure" else deadline
1392 stop_flag = threading.Event()
1393
1394 # Shared counters (only touched by the drain thread except stop_flag).
1395 # `errors`/`ramp_errors` count NON-throttle failures (a throttle is its own
1396 # signal, tallied separately and surfaced via the skew/taint path); a
1397 # non-throttle error is usually structural (ValidationException from a bad
1398 # index/attr/duplicate-key, AccessDenied, ResourceNotFound) and would
1399 # otherwise vanish — observed_cu is 0, so it can masquerade as a benign
1400 # expected-vs-observed delta. `error_codes` tallies the distinct codes so the
1401 # report can name the cause (e.g. {"ValidationException": 80}).
1402 stats: dict[ str , Any] = {
1403 "calls" : 0 ,
1404 "throttles" : 0 ,
1405 "ramp_calls" : 0 ,
1406 "ramp_throttles" : 0 ,
1407 "errors" : 0 ,
1408 "ramp_errors" : 0 ,
1409 "error_codes" : {},
1410 # Per-item Transact* cancellation reason histogram (e.g.
1411 # {"TransactionConflict": 13}). Tallied separately from error_codes
1412 # because the top-level code is always "TransactionCanceledException"
1413 # and hides whether the failures are contention (artifact) or a
1414 # genuine structural problem. Empty for non-transactional patterns.
1415 "cancellation_reason_codes" : {},
1416 }
1417
1418 # One DescribeTable on entry to warm SDK/TLS/credentials off the
1419 # critical path.
1420 try :
1421 ctx.client.describe_table( TableName = ctx.prefixed_table(pattern[ "table" ]))
1422 except ClientError:
1423 pass
1424
1425 pool = ThreadPoolExecutor( max_workers = ctx.concurrency)
1426 fut_queue: "queue.Queue" = queue.Queue()
1427
1428 def _drain_worker ():
1429 # Pulls (future, partition_idx) tuples off the queue, waits for each,
1430 # records the row. None sentinel means no more futures will be submitted.
1431 while True :
1432 item = fut_queue.get()
1433 if item is None :
1434 return
1435 fut_obj, part_idx = item
1436 try :
1437 res = fut_obj.result()
1438 except Exception as e:
1439 res = {
1440 "latency_ms" : 0.0 ,
1441 "consumed_cu" : 0.0 ,
1442 "gsi_cu" : {},
1443 "throttled" : False ,
1444 "error" : f "drain_exc: { type (e). __name__ } " ,
1445 }
1446 stats[ "calls" ] += 1
1447 in_ramp = time.monotonic() < ramp_deadline
1448 if res[ "throttled" ]:
1449 stats[ "throttles" ] += 1
1450 if in_ramp:
1451 stats[ "ramp_throttles" ] += 1
1452 elif res[ "error" ]:
1453 # Non-throttle failure — structural, not capacity. Count it
1454 # separately so it cannot hide inside a "0 observed CU" delta.
1455 stats[ "errors" ] += 1
1456 code = res[ "error" ]
1457 stats[ "error_codes" ][code] = stats[ "error_codes" ].get(code, 0 ) + 1
1458 # Unpack Transact* per-item cancellation reasons so the report
1459 # can distinguish contention (TransactionConflict) from a real
1460 # structural failure, instead of seeing only the generic
1461 # "TransactionCanceledException".
1462 for rc in res.get( "cancellation_reasons" ) or []:
1463 stats[ "cancellation_reason_codes" ][rc] = (
1464 stats[ "cancellation_reason_codes" ].get(rc, 0 ) + 1
1465 )
1466 if in_ramp:
1467 stats[ "ramp_errors" ] += 1
1468 if in_ramp:
1469 stats[ "ramp_calls" ] += 1
1470 sink.add(
1471 {
1472 "pattern_id" : pid,
1473 "op" : pattern[ "operation" ],
1474 "phase" : phase,
1475 "ts" : time.time(),
1476 "latency_ms" : res[ "latency_ms" ],
1477 "consumed_cu" : res[ "consumed_cu" ],
1478 "gsi_cu" : res[ "gsi_cu" ],
1479 # Vector capacity is metered in bytes, not CU — carried through
1480 # because a SearchVectors pattern has no other cost signal.
1481 "vector_search_bytes" : res.get( "vector_search_bytes" , 0.0 ),
1482 "vector_write_bytes" : res.get( "vector_write_bytes" ) or {},
1483 "throttled" : res[ "throttled" ],
1484 "error" : res[ "error" ],
1485 # Partition the read/write targeted — drives the per-pattern
1486 # key-distribution histogram (hot-partition skew, Mechanics #3).
1487 "key_idx" : part_idx,
1488 }
1489 )
1490 # Abort guard — active through ramp window during measurement.
1491 # `pid not in tainted` so a throttle burst doesn't overwrite an
1492 # error-rate taint already set below — the distinct reason strings
1493 # ("ramp:throttle_rate=" vs "error_rate=:CODE") must each survive so
1494 # the report can tell a throttled pattern from a structurally broken
1495 # one. (The downstream correctness finding keys off exact error
1496 # counts regardless, but the human-readable reason should be right.)
1497 if phase == "measure" and stats[ "ramp_calls" ] >= 20 and pid not in tainted:
1498 rate = stats[ "ramp_throttles" ] / max ( 1 , stats[ "ramp_calls" ])
1499 if rate > ctx.abort_throttle:
1500 tainted[pid] = f "ramp:throttle_rate= { rate :.2f} "
1501 stop_flag.set()
1502 elif phase == "warmup" and stats[ "throttles" ] > 0 and stats[ "calls" ] >= 20 :
1503 rate = stats[ "throttles" ] / max ( 1 , stats[ "calls" ])
1504 if rate > ctx.abort_throttle:
1505 # Warmup-only throttles: do NOT taint; stop warmup early.
1506 stop_flag.set()
1507 # Error-rate guard — distinct from throttling. A high NON-throttle
1508 # error rate during measurement means the pattern is structurally
1509 # broken (bad index/attr, duplicate key, access denied), not capacity-
1510 # bound: its observed CU/latency are meaningless, so taint it and stop
1511 # burning the window. The distinct taint reason ("error_rate=") lets
1512 # the report tell a broken pattern apart from a throttle-tainted one
1513 # and raise a correctness finding instead of a benign cost delta. The
1514 # error tally itself flows out via exact_counts regardless of taint.
1515 if phase == "measure" and stats[ "calls" ] >= 20 :
1516 erate = stats[ "errors" ] / max ( 1 , stats[ "calls" ])
1517 if erate > ctx.abort_error_rate and pid not in tainted:
1518 top_code = max (
1519 stats[ "error_codes" ], key = stats[ "error_codes" ].get, default = "error"
1520 )
1521 tainted[pid] = f "error_rate= { erate :.2f} : { top_code } "
1522 stop_flag.set()
1523
1524 drain_thread = threading.Thread( target = _drain_worker, daemon = True )
1525 drain_thread.start()
1526
1527 try :
1528 next_time = time.monotonic()
1529 key_counter = 0
1530 while time.monotonic() < deadline and not stop_flag.is_set():
1531 now = time.monotonic()
1532 if now < next_time:
1533 time.sleep( min ( 0.01 , next_time - now))
1534 continue
1535 next_time += interval
1536 # Draw the target partition here (once per call) so zipf's RNG draw
1537 # is counted exactly once and the drain can record which partition
1538 # was hit. member_counter cycles collection members within it.
1539 part_idx = ctx.sampled_idx(key_counter)
1540 fut = pool.submit(_dispatch, ctx, pattern, part_idx, key_counter)
1541 fut_queue.put((fut, part_idx))
1542 key_counter += 1
1543 finally :
1544 # Signal drain to stop after everything submitted has been processed.
1545 fut_queue.put( None )
1546 pool.shutdown( wait = True )
1547 drain_thread.join( timeout = 30 )
1548
1549 # Record EXACT (uncapped) tallies for this (pattern, phase) so throttle,
1550 # error, and call totals survive per-key row down-sampling.
1551 if counts is not None :
1552 counts[(pid, phase)] = {
1553 "calls" : stats[ "calls" ],
1554 "throttles" : stats[ "throttles" ],
1555 "errors" : stats[ "errors" ],
1556 "error_codes" : dict (stats[ "error_codes" ]),
1557 "cancellation_reason_codes" : dict (stats[ "cancellation_reason_codes" ]),
1558 }
1559
1560
1561 def run_warmup (ctx: RunContext, sink: RowSink, tainted: dict , counts: dict ):
1562 if ctx.warmup_seconds <= 0 :
1563 return
1564 # Serialize patterns so they don't compete in a tiny Lambda — but each
1565 # pattern uses its internal ThreadPoolExecutor so within-pattern calls
1566 # still overlap.
1567 for p in ctx.patterns:
1568 run_pattern_window(ctx, p, ctx.warmup_seconds, "warmup" , sink, tainted, counts)
1569
1570
1571 def run_measure (ctx: RunContext, sink: RowSink, tainted: dict , slice_seconds: float , counts: dict ):
1572 for p in ctx.patterns:
1573 if p[ "pattern_id" ] in tainted:
1574 continue
1575 run_pattern_window(ctx, p, slice_seconds, "measure" , sink, tainted, counts)
1576
1577
1578 # ---------------------------------------------------------------------------
1579 # Handler
1580 # ---------------------------------------------------------------------------
1581
1582
1583 def handler (event, context):
1584 started = time.monotonic()
1585 region = (event.get( "manifest" ) or {}).get( "region" ) or os.environ.get(
1586 "AWS_REGION" , "us-east-1"
1587 )
1588 cfg_block = event.get( "config" ) or {}
1589 # Tight client-side timeouts so one stuck call can't eat the budget.
1590 #
1591 # CRITICAL: data-path retries are DISABLED (max_attempts=1). boto3's default
1592 # "standard"/"legacy" retry modes transparently re-issue throttled requests
1593 # (ProvisionedThroughputExceeded / ThrottlingException) and only surface the
1594 # eventual success — which would make a load benchmark whose entire job is to
1595 # OBSERVE throttling report zero throttles even when the table is throttling
1596 # hard. We want each throttle counted once, not retried away. The orchestrator
1597 # already disables retries on the Lambda invoke for the same reason. (Seeding,
1598 # which is correctness-not-measurement, keeps its own UnprocessedItems retry
1599 # loop in run_seed.)
1600 # Size the HTTP connection pool to the per-pattern driver concurrency.
1601 # botocore defaults max_pool_connections to 10; the measurement driver runs
1602 # `concurrency_per_pattern` threads (default 32), so the default pool starves
1603 # — threads block waiting for a connection ("Connection pool is full"), which
1604 # inflates the measured p99 with DRIVER queueing that has nothing to do with
1605 # DynamoDB. Pool to the concurrency + headroom so the measured latency
1606 # reflects the service, not the client. (+8 covers the warmup warm-pool and
1607 # any incidental concurrent calls.)
1608 _pool = int (cfg_block.get( "concurrency_per_pattern" , 32 )) + 8
1609 client = boto3.client(
1610 "dynamodb" ,
1611 region_name = region,
1612 config = BotoConfig(
1613 retries = { "max_attempts" : 1 , "mode" : "standard" },
1614 connect_timeout = 3 ,
1615 read_timeout = 10 ,
1616 max_pool_connections = _pool,
1617 ),
1618 )
1619 ctx = RunContext(event, client)
1620 sink = RowSink( n_patterns = len (ctx.patterns) or 1 )
1621 tainted: dict = {}
1622 phases_run: list[ str ] = []
1623 seed_counts: dict = {}
1624 seed_verification: dict = {}
1625 # Exact (uncapped) per-(pattern, phase) call/throttle tallies. Survives the
1626 # per-key row cap so the summary's throttle/call counts are never truncated.
1627 exact_counts: dict = {}
1628
1629 try :
1630 if "settle" in ctx.phase_plan:
1631 run_settle(ctx, sink)
1632 phases_run.append( "settle" )
1633
1634 if "seed" in ctx.phase_plan:
1635 seed_counts = run_seed(ctx, sink)
1636 seed_verification = verify_seed(ctx, seed_counts)
1637 phases_run.append( "seed" )
1638 # Refuse to proceed if seeding manifestly failed (every target is empty).
1639 all_empty = seed_verification and all (
1640 v[ "actual" ] == 0 and v[ "expected" ] > 0 for v in seed_verification.values()
1641 )
1642 if all_empty:
1643 return {
1644 "invocation_index" : ctx.invocation_index,
1645 "phases_run" : phases_run,
1646 "raw_rows" : sink.drain(),
1647 "seed_verification" : seed_verification,
1648 "coverage" : {
1649 "measured_patterns" : [],
1650 "missing_patterns" : [p[ "pattern_id" ] for p in ctx.patterns],
1651 "coverage_incomplete" : True ,
1652 },
1653 "measurement_tainted" : tainted,
1654 "seed_verification_failed" : True ,
1655 "lambda_duration_seconds" : time.monotonic() - started,
1656 }
1657
1658 if "warmup" in ctx.phase_plan:
1659 run_warmup(ctx, sink, tainted, exact_counts)
1660 phases_run.append( "warmup" )
1661
1662 if "measure" in ctx.phase_plan:
1663 # Split the total measurement duration across invocations.
1664 slice_s = ctx.duration_seconds / max ( 1 , ctx.invocations_total)
1665 run_measure(ctx, sink, tainted, slice_s, exact_counts)
1666 phases_run.append( "measure" )
1667 except Exception as e:
1668 return {
1669 "invocation_index" : ctx.invocation_index,
1670 "phases_run" : phases_run,
1671 "raw_rows" : sink.drain(),
1672 "seed_verification" : seed_verification,
1673 "coverage" : {
1674 "measured_patterns" : [],
1675 "missing_patterns" : [],
1676 "coverage_incomplete" : True ,
1677 },
1678 "measurement_tainted" : tainted,
1679 "handler_error" : f " { type (e). __name__ } : { e } " ,
1680 "lambda_duration_seconds" : time.monotonic() - started,
1681 }
1682
1683 # Coverage check: every pattern must have produced at least one measure row
1684 # (only applies when measure phase was part of this invocation).
1685 rows = sink.drain()
1686 coverage = { "measured_patterns" : [], "missing_patterns" : [], "coverage_incomplete" : False }
1687 if "measure" in phases_run:
1688 measured = {r[ "pattern_id" ] for r in rows if r[ "phase" ] == "measure" }
1689 declared = {p[ "pattern_id" ] for p in ctx.patterns}
1690 coverage = {
1691 "measured_patterns" : sorted (measured),
1692 "missing_patterns" : sorted (declared - measured),
1693 "coverage_incomplete" : bool (declared - measured),
1694 }
1695
1696 return {
1697 "invocation_index" : ctx.invocation_index,
1698 "phases_run" : phases_run,
1699 "raw_rows" : rows,
1700 "seed_verification" : seed_verification,
1701 "coverage" : coverage,
1702 "measurement_tainted" : tainted,
1703 "lambda_duration_seconds" : time.monotonic() - started,
1704 # Exact per-(pattern, phase) call/throttle/error tallies (uncapped) so the
1705 # summary never under-reports throttles OR errors when rows were
1706 # down-sampled. error_codes names the distinct non-throttle failure codes.
1707 "exact_counts" : [
1708 {
1709 "pattern_id" : k[ 0 ],
1710 "phase" : k[ 1 ],
1711 "calls" : v[ "calls" ],
1712 "throttles" : v[ "throttles" ],
1713 "errors" : v.get( "errors" , 0 ),
1714 "error_codes" : v.get( "error_codes" , {}),
1715 "cancellation_reason_codes" : v.get( "cancellation_reason_codes" , {}),
1716 }
1717 for k, v in exact_counts.items()
1718 ],
1719 # Echo the load-shape knobs so the report can branch its disclaimer and
1720 # render the load-risk section only for representative-mode runs.
1721 "mode" : cfg_block.get( "mode" , "standard" ),
1722 "key_sampling" : (cfg_block.get( "read_pattern_key_sampling" ) or "uniform" ).lower(),
1723 "items_per_partition" : int (cfg_block.get( "items_per_partition" , 1 )),
1724 }