dm-limits-and-best-practices

# CDF Data Modeling: Limits, Concurrency & Best Practices This is a reference skill. When writing or reviewing code that calls CDF Data Modeling APIs, apply the patterns below. This skill owns runtime reliability concerns: limits, concurrency, retries, throughput, and batching behavior. For traversal payload correctness and graph-specific failure signatures, see `dm-graph-traversal`. --- ## DMS Limits Reference For the latest concurrency limits, resource limits, and property value limits, see the official documentation: **https://docs.cognite.com/cdf/dm/dm_reference/dm_limits_and_restrictions** Key things to be aware of: - Instance **apply**, **delete**, and **query** operations each have their own concurrent request limits - Exceeding these limits returns **429 Too Many Requests** - Transformations consume a large portion of the concurrency budget, leaving less for other clients - `instances.list` has a max page size (use pagination for complete results) - `instances.query` table expressions each have their own item limit - `instances.upsert` accepts up to 1000 items per call - `in` filters accept at most 1000 values per expression; larger sets must be split into batches --- ## Search vs Filter: When to Use Which ### `instances.search` — Free-text search on text properties Use `instances.search` when you need fuzzy/text matching on string fields (names, descriptions, etc.). It supports an `operator` parameter: - **`AND`** (default) — Narrow search. All terms must match. Use when the user provides a specific query. - **`OR`** — Broad "shotgun" search. Any term can match. Use for exploratory/typeahead search where you want maximum recall. ```typescript // Narrow search: find a specific cell by name (AND — all terms must match) const exactResults = await client.instances.search({ view: { type: 'view', ...PROCESS_CELL_VIEW }, query: 'reactor tank A', properties: ['name'], operator: 'AND', limit: 10, }); // Broad search: typeahead/autocomplete (OR — any term can match) const broadResults = await client.instances.search({ view: { type: 'view', ...BATCH_VIEW }, query: 'BUDE completed', properties: ['name', 'description', 'batchStatus'], operator: 'OR', limit: 10, }); ``` You can combine `search` with `filter` to further constrain results with exact-match conditions: ```typescript // Text search + exact filter: search for "pump" but only in active nodes const filtered = await client.instances.search({ view: { type: 'view', ...PROCESS_CELL_VIEW }, query: 'pump', properties: ['name', 'description'], filter: { equals: { property: getContainerProperty(MY_CONTAINER, 'status'), value: 'active', }, }, limit: 20, }); ``` ### `instances.list` / `instances.query` with `filter` — Exact-match filtering Use `filter` when you need precise, deterministic matching (equals, range, in, hasData, etc.). No fuzzy matching — values must match exactly. ```typescript // Exact match: get all completed batches const completedBatches = await client.instances.list({ instanceType: 'node', sources: [{ source: { type: 'view', ...BATCH_VIEW } }], filter: { equals: { property: getContainerProperty(BATCH_CONTAINER, 'batchStatus'), value: 'completed', }, }, limit: 1000, }); ``` ### Decision Guide | Need | Use | | ----------------------------------- | ----------------------------- | | User typing in a search box | `instances.search` with `OR` | | Find a specific item by name | `instances.search` with `AND` | | Filter by status, date range, enums | `filter` on list/query | | Text search + exact constraints | `instances.search` + `filter` | ### `in` filter value limit (1000) and batching CDF `in` filters support a maximum of 1000 values in a single filter expression. If you need to filter against more than 1000 IDs, split values into chunks and issue multiple requests, then merge results. ```typescript const IN_FILTER_BATCH_SIZE = 1000; // Reuse the Chunking Utility defined in the Batching Write Operations section. async function listByExternalIds( client: CogniteClient, externalIds: string[], ): Promise<NodeOrEdge[]> { const idBatches = chunk(externalIds, IN_FILTER_BATCH_SIZE); const responses = await Promise.all( idBatches.map((batch) => cdfTaskRunner.schedule(() => client.instances.list({ instanceType: 'node', sources: [{ source: { type: 'view', ...MY_VIEW } }], filter: { in: { property: ['node', 'externalId'], values: batch, }, }, limit: 1000, }) ) ) ); return responses.flatMap((r) => r.items); } ``` --- ## QueuedTaskRunner (Semaphore) **Always use the global `cdfTaskRunner`** to wrap CDF API calls. It limits concurrent requests and prevents 429 errors and deadlocks. ### Source Code If the project does not already have a semaphore utility, create `src/shared/utils/semaphore.ts` with this implementation: ```typescript /** * AbortError thrown when a queued task is cancelled */ export class AbortError extends Error { public constructor(message: string = 'Aborted') { super(message); this.name = 'AbortError'; } } type PendingTask<AsyncFn, AsyncFnResult> = { resolve: (result: AsyncFnResult) => void; reject: (error: unknown) => void; fn: AsyncFn; key?: string; }; const DEFAULT_MAX_CONCURRENT_TASKS = 15; /** * QueuedTaskRunner for controlling concurrent operations * Used to limit concurrent CDF API requests to avoid rate limiting and deadlocks * Essentially a semaphore that allows a limited number of tasks to run at once. */ export default class QueuedTaskRunner< AsyncFn extends () => Promise<AsyncFnResult>, AsyncFnResult = Awaited<ReturnType<AsyncFn>>, > { private pendingTasks: PendingTask<AsyncFn, AsyncFnResult>[] = []; private currentPendingTasks: number = 0; private readonly maxConcurrentTasks: number = 1; public constructor( maxConcurrentTasks: number = DEFAULT_MAX_CONCURRENT_TASKS ) { this.maxConcurrentTasks = maxConcurrentTasks; } public schedule( fn: AsyncFn, options: { key?: string } = {} ): Promise<AsyncFnResult> { this.startTrackingTime(); return new Promise((resolve, reject) => { if (options.key !== undefined) { // Cancel existing tasks with the same key (deduplication) this.pendingTasks .filter((task) => task.key === options.key) .forEach((task) => task.reject(new AbortError())); this.pendingTasks = this.pendingTasks.filter( (task) => task.key !== options.key ); } this.pendingTasks.push({ resolve, reject, fn, key: options.key, }); this.attemptConsumingNextTask(); }); } public async attemptConsumingNextTask(): Promise<void> { if (this.pendingTasks.length === 0) return; if (this.currentPendingTasks >= this.maxConcurrentTasks) return; const pendingTask = this.pendingTasks.shift(); if (pendingTask === undefined) { throw new Error('pendingTask is undefined, this should never happen'); } this.currentPendingTasks++; const { fn, resolve, reject } = pendingTask; try { const result = await fn(); resolve(result); } catch (e) { reject(e); } finally { this.currentPendingTasks--; this.tick(); this.attemptConsumingNextTask(); } } public clearQueue = (): void => { this.pendingTasks = []; }; private startTime: number | null = null; private startTrackingTime = (): void => { if (this.startTime === null) { this.startTime = performance.now(); } }; private tick = (): void => { if (this.pendingTasks.length === 0) { this.startTime = null; } }; } /** * Global task runner for CDF API requests * Limits concurrent requests to avoid 429 rate limiting and deadlocks */ export const cdfTaskRunner = new QueuedTaskRunner(DEFAULT_MAX_CONCURRENT_TASKS); ``` ### Usage Pattern Always wrap CDF calls with `cdfTaskRunner.schedule()`: ```typescript import { cdfTaskRunner } from '../../../../shared/utils/semaphore'; // Single query export async function fetchBatches(client: CogniteClient): Promise<CDFBatch[]> { return cdfTaskRunner.schedule(async () => { const response = await client.instances.query({ with: { /* ... */ }, select: { /* ... */ }, }); return response.items?.nodes || []; }); } // Multiple parallel queries (safe — the semaphore limits concurrency) export async function enrichBatch( client: CogniteClient, batch: CDFBatch ): Promise<BatchEnrichment> { const [currentOp, lastOp, cells, material] = await Promise.all([ fetchCurrentOperation(client, batch.space, batch.externalId), fetchLastCompletedOperation(client, batch.space, batch.externalId), fetchProcessCells(client, batch.space, batch.externalId), fetchMaterial(client, batch.space, batch.externalId), ]); return { currentOp, lastOp, cells, material }; } // Each of the above functions internally uses cdfTaskRunner.schedule(), // so Promise.all is safe — the semaphore prevents exceeding concurrency limits ``` ### Deduplication with Keys Use the `key` option to cancel stale requests when the same query is triggered again (e.g., user changes filters quickly): ```typescript const result = await cdfTaskRunner.schedule( async () => client.instances.query({ /* ... */ }), { key: `batch-flow-${batchId}` } ); // If another call with the same key arrives before this completes, // the previous pending call is rejected with AbortError ``` --- ## Pagination DMS `instances.list` returns at most `limit` items and a `nextCursor` for the next page. DMS `instances.query` uses a `cursors` object keyed by table expression name. ### instances.list Pagination ```typescript async function fetchAllNodes(client: CogniteClient): Promise<CDFNodeResponse[]> { const allItems: CDFNodeResponse[] = []; let cursor: string | undefined = undefined; do { const response = await client.instances.list({ instanceType: 'node', sources: [{ source: { type: 'view', ...MY_VIEW } }], filter: { equals: { property: getContainerProperty(MY_CONTAINER, 'status'), value: 'active', }, }, limit: 1000, cursor, }); allItems.push(...response.items); cursor = response.nextCursor; } while (cursor); return allItems; } ``` ### instances.query Pagination The `query` endpoint returns `nextCursor` as a `Record<string, string>` (one cursor per table expression). Use it via the `cursors` parameter: ```typescript import { isEmpty } from 'lodash'; async function fetchAllResults( client: CogniteClient ): Promise<{ results: CDFResult[]; edges: EdgeDefinition[] }> { const QUERY_LIMIT = 10_000; const fetchPage = async ( nextCursors?: Record<string, string> ): Promise<{ results: CDFResult[]; edges: EdgeDefinition[] }> => { const { items, nextCursor } = await client.instances.query({ with: { results: { limit: QUERY_LIMIT, nodes: { filter: { hasData: [{ type: 'view', ...RESULT_VIEW }], }, }, }, relatedEdges: { limit: QUERY_LIMIT, edges: { from: 'results' as const, maxDistance: 1, direction: 'outwards' as const, filter: { equals: { property: ['edge', 'type'], value: MY_EDGE_TYPE, }, }, }, }, }, cursors: nextCursors, // Pass cursors from previous page select: { results: { sources: [ { source: { type: 'view', ...RESULT_VIEW }, properties: ['*'] }, ], }, relatedEdges: {}, }, }); const results = (items?.results || []) as CDFResult[]; const edges = (items?.relatedEdges || []).filter( (e) => e.instanceType === 'edge' );