How Asset Collection Works in OpenMAIC with PostgreSQL Persistence

OpenMAIC's asset collection mechanism periodically deletes unreferenced binary blobs from PostgreSQL by computing a grace-period cutoff, batching candidates, and executing atomic transactions that lock rows, verify zero references, delete bytes, and remove metadata rows.

OpenMAIC stores binary assets—such as uploaded images, audio, and video—in a PostgreSQL-backed persistence layer. To prevent infinite storage growth, the system implements a garbage collection process that reclaims orphaned blobs after a configurable safety window, all while maintaining strict transactional consistency.

The Asset Storage Architecture

The asset_blobs Table

Binary assets reside in the asset_blobs table. Each row tracks a blob’s content_hash (its unique identifier) and an unreferenced_at timestamp indicating when the last reference to that blob was removed. Only rows with an unreferenced_at value older than the configured grace period are eligible for deletion.

The AssetCollector Class

The core logic lives in AssetCollector (packages/@openmaic/storage/src/asset/collector.ts). This class orchestrates bounded collection passes using two dependencies:

• Queryable: A thin PostgreSQL query surface (typically a pg.Pool or similar) for executing SQL.
• AssetByteStore: An interface (packages/@openmaic/storage/src/asset/byte-store.ts) that handles the actual deletion of raw bytes, optionally providing a deleteWith method to avoid self-deadlocks.

Step-by-Step Collection Process

The collectPass() method implements a six-stage workflow:

1. Initialization and Cutoff Calculation

When instantiated, the collector accepts options for graceMs, batchSize, and a clock override for testing. At runtime, it computes a cutoff timestamp using now() – graceMs. Only blobs whose unreferenced_at predates this cutoff are considered candidates.

See the cutoff calculation in packages/@openmaic/storage/src/asset/collector.ts#L68.

2. Candidate Fetching

The collector queries the asset_blobs table with a bounded SELECT statement limited to batchSize rows. Results are ordered by unreferenced_at (oldest first) and then by content_hash to ensure deterministic, starvation-free processing.

This query appears at packages/@openmaic/storage/src/asset/collector.ts#L83.

3. Transactional Processing

For each candidate, the collector opens a withTransaction block to guarantee atomicity. Inside the transaction:

• Row locking: The candidate row is locked with FOR UPDATE to prevent concurrent modifications.
• Reference verification: A NOT EXISTS check confirms no entries in the asset_entries table still reference this blob.
• Byte deletion: The AssetByteStore deletes the raw bytes. If the store implements deleteWith, the deletion uses the same Queryable to avoid self-deadlock; otherwise, the bytes must live outside the same database.
• Metadata cleanup: The asset_blobs row is deleted.

This transaction body is implemented at packages/@openmaic/storage/src/asset/collector.ts#L96-L124.

4. Pass Result Reporting

After iterating through candidates, collectPass() returns an AssetCollectionPass object containing:

• collected: The count of blobs actually deleted during this pass.
• capped: A boolean indicating whether the result hit the batchSize limit. When true, additional passes are required to clear remaining eligible blobs.

The return statement is located at packages/@openmaic/storage/src/asset/collector.ts#L129-L130.

5. Scheduling and Idempotency

Hosts invoke collectPass() on a recurring schedule (e.g., via a cron job every few minutes). The design is idempotent; if multiple hosts run simultaneously, PostgreSQL's row-level locks ensure only one process deletes any specific blob.

Configuration and Safety Guarantees

OpenMAIC enforces several safety mechanisms to protect data integrity and system stability:

• Grace period: The default DEFAULT_ASSET_COLLECTION_GRACE_MS is one hour. This ensures signed URLs (which must expire earlier) remain valid while the underlying bytes are protected from premature deletion. The relationship is validated by assertSignedUrlTtlWithinGrace.
• Batch size: DEFAULT_ASSET_COLLECTION_BATCH_SIZE defaults to 1000, limiting transaction scope and preventing long-running queries from impacting database performance.
• Self-deadlock protection: If an AssetByteStore lacks the deleteWith method but stores bytes within the same PostgreSQL database, the collector throws a collectorConfigurationFailure() error. This prevents a deadlock scenario where the delete operation would wait on a row lock already held by the transaction.
• Transactional safety: All read operations, reference checks, byte deletions, and row removals occur within a single PostgreSQL transaction, ensuring atomic rollback if any step fails.

Implementation Example

The following TypeScript example demonstrates wiring the collector with a PostgreSQL pool and a byte store:

import { AssetCollector } from '@openmaic/storage/src/asset/collector';
import { PgAssetByteStore } from '@openmaic/storage/src/asset/byte-store';
import pg from 'pg';

// 1️⃣ Create a Queryable and transaction helper
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const withTransaction = async <T>(fn: (q: any) => Promise<T>) => {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const result = await fn(client);
    await client.query('COMMIT');
    return result;
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
};

// 2️⃣ Instantiate the byte store and collector
const byteStore = new PgAssetByteStore(/* configuration */);
const collector = new AssetCollector(pool, byteStore, {
  withTransaction,
  graceMs: 2 * 60 * 60 * 1000, // 2 hours
  batchSize: 500,
});

// 3️⃣ Run a bounded collection pass (typically from a background worker)
async function runAssetCollection() {
  const pass = await collector.collectPass();
  console.log(`Deleted ${pass.collected} unreferenced assets`);
  
  if (pass.capped) {
    // More work remains—schedule another pass immediately
    setTimeout(runAssetCollection, 0);
  }
}

runAssetCollection();

Summary

• AssetCollector (packages/@openmaic/storage/src/asset/collector.ts) manages the lifecycle of unreferenced binary blobs in PostgreSQL.
• Blobs become eligible for deletion only after their unreferenced_at timestamp exceeds the configurable graceMs window.
• Each collection pass processes a bounded batch (default 1000) within atomic transactions that lock rows, verify zero references, delete bytes, and remove metadata.
• The idempotent design allows safe concurrent execution across multiple hosts without double-deletion risks.
• Built-in safeguards prevent self-deadlocks and ensure signed URLs expire safely before their underlying assets are reclaimed.

Frequently Asked Questions

What is the default grace period for asset collection?

The default grace period is one hour (DEFAULT_ASSET_COLLECTION_GRACE_MS). This duration ensures that presigned URLs—which must expire earlier than the grace period—remain functional while protecting recently unreferenced assets from premature deletion.

How does the collector prevent database deadlocks?

The collector prevents deadlocks through two mechanisms. First, it uses FOR UPDATE row locks within transactions to serialize access to candidate blobs. Second, it validates that the AssetByteStore implements a deleteWith method when bytes reside in the same PostgreSQL instance; if this contract is violated, the collector throws a configuration error rather than risk a circular lock dependency.

Can multiple collector instances run simultaneously?

Yes. The architecture is explicitly idempotent and supports concurrent execution. Because each candidate row is locked with FOR UPDATE before deletion, only one process can successfully delete a specific blob even if multiple hosts invoke collectPass() at the same time. Failed attempts simply skip to the next candidate.

How do I adjust the batch size for collection passes?

Pass a batchSize option when instantiating AssetCollector. The default is 1000 rows per pass. Lower values reduce transaction duration and memory pressure but require more frequent passes to clear large backlogs of unreferenced assets.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}