# How Asset Collection Works in OpenMAIC with PostgreSQL Persistence

> Discover how OpenMAIC asset collection with PostgreSQL persistence automatically deletes unreferenced binary blobs using atomic transactions and efficient batching for optimized storage.

- Repository: [MAIC/OpenMAIC](https://github.com/THU-MAIC/OpenMAIC)
- Tags: how-to-guide
- Published: 2026-08-31

---

**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`](https://github.com/THU-MAIC/OpenMAIC/blob/main/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`](https://github.com/THU-MAIC/OpenMAIC/blob/main/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`](https://github.com/THU-MAIC/OpenMAIC/blob/main/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`](https://github.com/THU-MAIC/OpenMAIC/blob/main/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:

```typescript
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.