Skip to content

Commit f629fde

Browse files
pyramationpyramation
committed
refactor: unified usage logging via MetaSchema-resolved table names
- All metering (compute/inference/storage) now resolves table names from MetaSchema module registration tables instead of hardcoding schema names - Each meter uses a 60s TTL cache for resolved table references - Falls back to well-known defaults when MetaSchema is unavailable - fn-runtime/storage.ts: self-contained MetaSchema resolution for S3 metering - inference-meter.ts: self-contained MetaSchema resolution for LLM metering - storage-meter.ts in worker: delegates to UsageClient (same directory) - Updated storage-meter tests to account for MetaSchema lookup queries
1 parent 03c4c3a commit f629fde

5 files changed

Lines changed: 202 additions & 80 deletions

File tree

‎job/worker/src/storage-meter.ts‎

Lines changed: 17 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,58 +1,36 @@
11
/**
22
* Storage metering — fire-and-forget usage logging for S3/MinIO operations.
33
*
4-
* Logs to `constructive_usage_public.platform_usage_log_storage`
5-
* after each read/write/delete against object storage.
4+
* Delegates to UsageClient which resolves table names dynamically from
5+
* MetaSchema module registration tables. Schema-qualified references
6+
* are discovered at runtime, not hardcoded.
67
*
78
* All writes are non-blocking: errors are logged and swallowed so
89
* metering never affects function execution or storage latency.
910
*/
1011

11-
import { Logger } from '@pgpmjs/logger';
1212
import type { Pool } from 'pg';
13-
import { randomUUID } from 'crypto';
1413

15-
const log = new Logger('storage-meter');
14+
import type { StorageEntry } from './usage-client';
15+
import { UsageClient } from './usage-client';
1616

17-
export interface StorageEntry {
18-
databaseId?: string;
19-
entityId?: string;
20-
actorId?: string;
21-
operation: 'read' | 'write' | 'delete';
22-
bucket: string;
23-
key: string;
24-
sizeBytes: number;
25-
durationMs: number;
17+
export type { StorageEntry };
18+
19+
/** Module-level client cache — reused across calls within the same pool. */
20+
let _client: UsageClient | null = null;
21+
let _pool: Pool | null = null;
22+
23+
function getOrCreateClient(pool: Pool): UsageClient {
24+
if (_client && _pool === pool) return _client;
25+
_client = new UsageClient(pool);
26+
_pool = pool;
27+
return _client;
2628
}
2729

2830
/**
2931
* Log a storage operation to the usage log table.
3032
* Fire-and-forget: returns immediately, never throws.
3133
*/
3234
export function logStorageUsage(pool: Pool, entry: StorageEntry): void {
33-
const id = randomUUID();
34-
const now = new Date();
35-
36-
pool
37-
.query(
38-
`INSERT INTO "constructive_usage_public".platform_usage_log_storage
39-
(id, database_id, entity_id, actor_id, operation,
40-
bucket, key, size_bytes, duration_ms, created_at)
41-
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
42-
[
43-
id,
44-
entry.databaseId ?? null,
45-
entry.entityId ?? null,
46-
entry.actorId ?? null,
47-
entry.operation,
48-
entry.bucket,
49-
entry.key,
50-
entry.sizeBytes,
51-
Math.round(entry.durationMs),
52-
now
53-
]
54-
)
55-
.catch((err) => {
56-
log.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
57-
});
35+
getOrCreateClient(pool).logStorageUsage(entry);
5836
}

‎job/worker/src/usage-client.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ const KNOWN_TABLES = [
3535
'platform_function_invocations',
3636
'platform_usage_log_computes',
3737
'platform_usage_log_inferences',
38+
'platform_usage_log_storage',
3839
] as const;
3940

4041
/** Default config used when metaschema is unavailable */
@@ -45,6 +46,8 @@ const DEFAULTS: UsageModuleConfig = {
4546
computeUsageTable: 'platform_usage_log_computes',
4647
inferenceUsageSchema: 'constructive_usage_public',
4748
inferenceUsageTable: 'platform_usage_log_inferences',
49+
storageUsageSchema: 'constructive_usage_public',
50+
storageUsageTable: 'platform_usage_log_storage',
4851
};
4952

5053
export interface UsageModuleConfig {
@@ -54,6 +57,8 @@ export interface UsageModuleConfig {
5457
computeUsageTable: string;
5558
inferenceUsageSchema: string;
5659
inferenceUsageTable: string;
60+
storageUsageSchema: string;
61+
storageUsageTable: string;
5762
}
5863

5964
// ─── Entry Types ──────────────────────────────────────────────────────────────
@@ -92,6 +97,17 @@ export interface InferenceEntry {
9297
rawUsage?: unknown;
9398
}
9499

100+
export interface StorageEntry {
101+
databaseId?: string;
102+
entityId?: string;
103+
actorId?: string;
104+
operation: 'read' | 'write' | 'delete';
105+
bucket: string;
106+
key: string;
107+
sizeBytes: number;
108+
durationMs: number;
109+
}
110+
95111
// ─── UsageClient ──────────────────────────────────────────────────────────────
96112

97113
const DEFAULT_DATABASE_ID = '00000000-0000-0000-0000-000000000000';
@@ -138,6 +154,9 @@ export class UsageClient {
138154
if (row.table_name === 'platform_usage_log_inferences') {
139155
config.inferenceUsageSchema = row.schema_name;
140156
}
157+
if (row.table_name === 'platform_usage_log_storage') {
158+
config.storageUsageSchema = row.schema_name;
159+
}
141160
}
142161

143162
this.config = config;
@@ -284,6 +303,48 @@ export class UsageClient {
284303
log.warn(`inference log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
285304
});
286305
}
306+
307+
// ─── Storage Usage ─────────────────────────────────────────────────────
308+
309+
/**
310+
* Log a storage operation to the usage log table.
311+
* Fire-and-forget: returns immediately, never throws.
312+
*/
313+
logStorageUsage(entry: StorageEntry): void {
314+
this.resolveConfig()
315+
.then((cfg) => this.insertStorageUsage(cfg, entry))
316+
.catch((err) => {
317+
log.warn(`storage usage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
318+
});
319+
}
320+
321+
private async insertStorageUsage(cfg: UsageModuleConfig, entry: StorageEntry): Promise<void> {
322+
const id = randomUUID();
323+
const now = new Date();
324+
325+
this.pool
326+
.query(
327+
`INSERT INTO "${cfg.storageUsageSchema}"."${cfg.storageUsageTable}"
328+
(id, database_id, entity_id, actor_id, operation,
329+
bucket, key, size_bytes, duration_ms, created_at)
330+
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
331+
[
332+
id,
333+
entry.databaseId ?? null,
334+
entry.entityId ?? null,
335+
entry.actorId ?? null,
336+
entry.operation,
337+
entry.bucket,
338+
entry.key,
339+
entry.sizeBytes,
340+
Math.round(entry.durationMs),
341+
now,
342+
]
343+
)
344+
.catch((err) => {
345+
log.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
346+
});
347+
}
287348
}
288349

289350
/**

‎packages/agentic-server/src/inference-meter.ts‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,8 @@
11
/**
22
* Inference metering — fire-and-forget usage logging for LLM calls.
33
*
4-
* Self-contained implementation that resolves table names dynamically
5-
* from MetaSchema module registration tables. Falls back to well-known
6-
* defaults when MetaSchema is unavailable.
4+
* Resolves table names dynamically from MetaSchema module registration tables.
5+
* Falls back to well-known defaults when MetaSchema is unavailable.
76
*
87
* All writes are non-blocking: errors are logged and swallowed so
98
* metering never affects inference latency or response delivery.
@@ -35,7 +34,7 @@ export interface InferenceEntry {
3534
rawUsage?: unknown;
3635
}
3736

38-
// ─── Table Resolution ─────────────────────────────────────────────────────────
37+
// ─── Table Resolution (MetaSchema module loader) ──────────────────────────────
3938

4039
const INFERENCE_TABLE_SQL = `
4140
SELECT s.schema_name, t.name AS table_name

‎packages/fn-runtime/src/storage.ts‎

Lines changed: 71 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,53 @@ export type StorageMeterCallback = (info: {
2121
durationMs: number;
2222
}) => void;
2323

24+
// ─── MetaSchema Table Resolution ─────────────────────────────────────────────
25+
26+
const STORAGE_TABLE_SQL = `
27+
SELECT s.schema_name, t.name AS table_name
28+
FROM metaschema_public."table" t
29+
JOIN metaschema_public.schema s ON t.schema_id = s.id
30+
WHERE t.database_id = $1
31+
AND t.name = 'platform_usage_log_storage'
32+
LIMIT 1
33+
`;
34+
35+
const DEFAULT_DATABASE_ID = '00000000-0000-0000-0000-000000000000';
36+
const DEFAULT_SCHEMA = 'constructive_usage_public';
37+
const DEFAULT_TABLE = 'platform_usage_log_storage';
38+
const CONFIG_TTL_MS = 60_000;
39+
40+
let _storageSchema = DEFAULT_SCHEMA;
41+
let _storageTable = DEFAULT_TABLE;
42+
let _cacheExpiresAt = 0;
43+
let _resolvePromise: Promise<void> | null = null;
44+
45+
async function resolveStorageTable(pool: import('pg').Pool): Promise<void> {
46+
if (Date.now() < _cacheExpiresAt) return;
47+
if (_resolvePromise) return _resolvePromise;
48+
49+
_resolvePromise = (async () => {
50+
try {
51+
const { rows } = await pool.query(STORAGE_TABLE_SQL, [DEFAULT_DATABASE_ID]);
52+
if (rows.length > 0) {
53+
_storageSchema = rows[0].schema_name;
54+
_storageTable = rows[0].table_name;
55+
}
56+
} catch {
57+
meterLog.debug('metaschema lookup unavailable — using default storage table names');
58+
}
59+
_cacheExpiresAt = Date.now() + CONFIG_TTL_MS;
60+
_resolvePromise = null;
61+
})();
62+
63+
return _resolvePromise;
64+
}
65+
2466
/**
2567
* Create a fire-and-forget storage metering callback backed by pg.
2668
*
2769
* Lazily creates a pg Pool from standard PG* env vars on first invocation.
70+
* Resolves table names dynamically from MetaSchema (scope-aware).
2871
* Returns undefined if PGHOST/DATABASE_URL is not set (metering disabled).
2972
*/
3073
export const createMeterCallback = (): StorageMeterCallback | undefined => {
@@ -43,28 +86,34 @@ export const createMeterCallback = (): StorageMeterCallback | undefined => {
4386

4487
return (info) => {
4588
const p = getPool();
46-
const id = randomUUID();
47-
const now = new Date();
48-
p.query(
49-
`INSERT INTO "constructive_usage_public".platform_usage_log_storage
50-
(id, database_id, entity_id, actor_id, operation,
51-
bucket, key, size_bytes, duration_ms, created_at)
52-
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
53-
[
54-
id,
55-
info.databaseId ?? null,
56-
info.entityId ?? null,
57-
info.actorId ?? null,
58-
info.operation,
59-
info.bucket,
60-
info.key,
61-
info.sizeBytes,
62-
Math.round(info.durationMs),
63-
now
64-
]
65-
).catch((err: unknown) => {
66-
meterLog.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
67-
});
89+
resolveStorageTable(p)
90+
.then(() => {
91+
const id = randomUUID();
92+
const now = new Date();
93+
p.query(
94+
`INSERT INTO "${_storageSchema}"."${_storageTable}"
95+
(id, database_id, entity_id, actor_id, operation,
96+
bucket, key, size_bytes, duration_ms, created_at)
97+
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
98+
[
99+
id,
100+
info.databaseId ?? null,
101+
info.entityId ?? null,
102+
info.actorId ?? null,
103+
info.operation,
104+
info.bucket,
105+
info.key,
106+
info.sizeBytes,
107+
Math.round(info.durationMs),
108+
now
109+
]
110+
).catch((err: unknown) => {
111+
meterLog.warn(`storage log failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
112+
});
113+
})
114+
.catch((err: unknown) => {
115+
meterLog.warn(`storage resolve failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
116+
});
68117
};
69118
};
70119

0 commit comments

Comments
 (0)