# Stream an export and clean up failure
> Write incrementally, publish only a completed file, and leave the last successful export available when generation fails.
- Package: `@db3.ai/app/storage`
- Canonical page: [https://db3.ai/docs/cookbook-streams](https://db3.ai/docs/cookbook-streams)
- Markdown: [https://db3.ai/docs/cookbook-streams.md](https://db3.ai/docs/cookbook-streams.md)
- Framework source of truth: `packages/app/src/storage/README.md`
## Set up
Use the standalone app and development tools from Installation. This example needs only Node.js and temporary local storage. It does not open a database or contact a provider.
- [Install the tools](https://db3.ai/docs/installation.md)
## Copy the export
The writer accepts an already-selected disk and authorized destination. The runner owns its temporary directory and removes it in `finally`.
### Copy the shipped example
```bash
mkdir -p examples
cp -R node_modules/@db3.ai/app/src/storage/examples/. examples/
```
## Run it
Expect `lines: 50001` (header plus 50,000 rows), `pendingFiles: 0`, and true values for completion, interruption rejection, preserving the previous file and recovery. Counting the returned stream does not retain the whole export.
### Run the lab
```bash
npx tsx examples/runStreamExport.ts
```
## Stage before publishing
`Readable.from()` produces one row at a time. `writeStream()` consumes the stream before `move()` publishes it. A unique staging path prevents a failed producer from truncating the previous destination.
The catch path destroys the source and removes its partial staging object. Storage may wrap the original stream error in `cause`. If cleanup itself fails, surface that failure and reconcile the owned temporary path; do not pretend the operation completed.
These synthetic CSV values are safe. Real exports must escape delimiters, quotes and newlines, and consider spreadsheet formula injection when handling untrusted cells.
### examples/writeCsvExport.ts
```typescript
import { randomUUID } from 'node:crypto';
import { Readable } from 'node:stream';
import type { StorageDisk } from '@db3.ai/app/storage';
/** Options for a bounded-memory CSV export and deliberate failure in the lab. */
export interface CsvExportOptions {
/** Number of synthetic rows to generate without retaining all rows in memory. */
rows: number;
/** Lab-only failure point; omit for a normal export. */
failAt?: number;
}
/**
* Streams a generated CSV to a staging path, publishing only after success.
*
* The staging path is owned by this operation. A failed stream removes that
* object and leaves the previous destination untouched. Cross-provider moves
* are not assumed atomic; concurrent publication needs an application policy.
*
* @param disk - Application-owned disk selected before the operation.
* @param destination - Authorized relative output path, not raw request input.
* @param options - Row count and optional controlled failure point.
*/
export async function writeCsvExport(disk: StorageDisk, destination: string, options: CsvExportOptions): Promise {
if (!Number.isInteger(options.rows) || options.rows < 0 || options.rows > 1_000_000) throw new Error('rows must be an integer between 0 and 1000000');
const staging = `pending/${randomUUID()}.csv`;
const stream = Readable.from(csvRows(options));
try {
await disk.writeStream(staging, stream, { mimeType: 'text/csv', visibility: 'private' });
await disk.move(staging, destination);
} catch (error) {
stream.destroy();
if (await disk.exists(staging)) await disk.delete(staging);
throw error;
}
}
/** Produces one safe synthetic CSV row at a time; real exports must escape data. */
function* csvRows(options: CsvExportOptions): Generator {
yield 'id,value\n';
for (let index = 0; index < options.rows; index++) {
if (index === options.failAt) throw new Error('Controlled export interruption');
yield `${index},note-${index}\n`;
}
}
```
## Consume without buffering
Use `pipeline()` to connect the readable to a writable and propagate errors. Avoid `readToBuffer()` or `readToString()` when the file size is unbounded. The runner counts bytes/newlines incrementally and compares the byte count with storage metadata.
### examples/runStreamExport.ts
```typescript
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { pathToFileURL } from 'node:url';
import { Storage } from '@db3.ai/app/storage';
import { writeCsvExport } from './writeCsvExport';
/**
* Exercises streamed file publication, interrupted-write cleanup and retry.
* @returns Counts and cleanup outcomes without buffering the complete export.
*/
export async function runStreamExport() {
const root = await mkdtemp(join(tmpdir(), 'db3-stream-guide-'));
const disk = new Storage({ disks: { local: { driver: 'local', root } } }).disk('local');
try {
await writeCsvExport(disk, 'exports/latest.csv', { rows: 50_000 });
const sizeBefore = await disk.size('exports/latest.csv');
let interruptionRejected = false;
try { await writeCsvExport(disk, 'exports/latest.csv', { rows: 10, failAt: 2 }); } catch (error) {
// Drivers may wrap failures while preserving the original stream error.
const cause = error instanceof Error && error.cause instanceof Error ? error.cause : error;
if (!(cause instanceof Error) || cause.message !== 'Controlled export interruption') throw error;
interruptionRejected = true;
}
const previousPreserved = await disk.size('exports/latest.csv') === sizeBefore;
const pendingFiles = (await disk.list('pending').toArray()).filter(entry => entry.isFile).length;
await writeCsvExport(disk, 'exports/repaired.csv', { rows: 3 });
let lines = 0;
let bytes = 0;
const sink = new Writable({
/** Counts each chunk without retaining the complete file. */
write(chunk: Buffer, _encoding, callback) { bytes += chunk.length; for (const byte of chunk) if (byte === 10) lines++; callback(); },
});
await pipeline(await disk.readStream('exports/latest.csv'), sink);
return { lines, complete: bytes === sizeBefore, interruptionRejected, previousPreserved, pendingFiles, repaired: await disk.exists('exports/repaired.csv') };
} finally { await rm(root, { recursive: true, force: true }); }
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) console.log(JSON.stringify(await runStreamExport(), null, 2));
```
## Know what the stream does not solve
Local rename and remote object moves do not share an atomicity guarantee. Serialize competing publishers or use immutable versioned destinations plus an application-owned pointer. This test does not certify an S3 provider.
An HTTP adapter must authorize before streaming and stop work on client disconnect. A queue export should be retry-safe and avoid publishing duplicate side effects. This lab proves producer failure cleanup, not network cancellation or crash recovery.
- [Private HTTP download](https://db3.ai/docs/guide-files.md)
- [Background work](https://db3.ai/docs/guide-background.md)
## Testing
Create a `tests/storage` directory and save the test below as `runStreamExport.test.ts`. It imports the example you copied into `examples/`. Keep the same folder layout so the relative import resolves.
Run from the application root with the development dependencies from Installation. These are consumer tests, not commands that assume a framework checkout.
### tests/storage/runStreamExport.test.ts
```typescript
import { expect, it } from 'vitest';
import { runStreamExport } from '../../examples/runStreamExport';
it('streams a large export, preserves the previous file on interruption and cleans partial bytes', async () => {
expect(await runStreamExport()).toEqual({ lines: 50_001, complete: true, interruptionRejected: true, previousPreserved: true, pendingFiles: 0, repaired: true });
});
```
## Run and extend the test
Change the row count and move the controlled failure to another row. Keep the old-destination and empty-staging assertions. Test your real driver with a dedicated bucket before relying on its move behavior.
### Run your copied test and check types
```bash
npx vitest run tests/storage/runStreamExport.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.ts
```
## Coverage
Tested: real local streaming, complete output, producer failure, previous-file preservation, partial cleanup and retry. Still to build: a socket-disconnect trial, resumable uploads and remote-driver failure/concurrency conformance.
- [Complete Storage API](https://db3.ai/docs/storage-api.md)
## Behavioural verification
Streams 50,000 CSV rows, interrupts a staged write, preserves the previous export, removes partial bytes and retries.
- Behaviour test: `packages/app/src/storage/tests/examples/runStreamExport.test.ts`
- Repository test command (framework checkout only): `npm run test:service --workspace @platform/app -- storage --maxWorkers=1`
- In an installed application, use the walkthrough commands instead of this repository test.
- Expected outcome: The guide test passes against the real framework components.
- Environment: Temporary local filesystem, no SQL, HTTP socket or remote provider.
## Related documentation
- [Storage](https://db3.ai/docs/storage.md): Write files to a named disk, stream large payloads and keep paths relative to storage. Add Media when files need durable identities and ownership metadata.
- [Upload and download private files](https://db3.ai/docs/guide-files.md): A file ID is not permission to read it. Start with a small, authenticated text-file endpoint and keep the policy visible.
- [Queue](https://db3.ai/docs/queue-overview.md): Create durable background jobs, run named workers, understand every attempt, compose chains and batches, and recover failures through one complete service guide.
## Framework-owned source: `packages/app/src/storage/README.md`
This is the exact source document captured by the documentation build. Use it for detailed API and workflow guidance, subject to the public package exports and behavioural evidence identified above.
````markdown
# @db3.ai/app/storage
## Stream failure and recovery lab
Copy `src/storage/examples/` from the installed package into `examples/` and run
`npx tsx examples/runStreamExport.ts`. The lab streams 50,000 rows, interrupts a
staged write, preserves the last successful export and removes partial bytes.
It needs only temporary local storage and removes that directory afterwards.
`tests/examples/runStreamExport.test.ts` is the consumer-copyable behavioral
check. Remote moves and HTTP disconnects need separate adapter/driver tests.
`@db3.ai/app/storage` provides a Laravel-style storage service with named
disks. App code can write to the default disk through `app.storage` or select a
configured disk with `app.storage.disk(name)`.
The storage service is the lower-level file API. Managed file records, ULID
references, and database-backed file metadata should sit above this layer.
## Configure disks
```ts
import { App } from '@db3.ai/app';
const app = new App({
storage: {
default: 'local',
disks: {
local: {
driver: 'local',
root: 'storage/app',
},
agent: {
driver: 'local',
root: 'storage/agent',
url: 'https://example.test/storage/agent',
},
spaces: {
driver: 's3',
bucket: 'example-assets',
region: 'nyc3',
endpoint: 'https://nyc3.digitaloceanspaces.com',
url: 'https://example-assets.nyc3.digitaloceanspaces.com',
accessKeyId: process.env.SPACES_ACCESS_KEY_ID,
secretAccessKey: process.env.SPACES_SECRET_ACCESS_KEY,
},
},
},
});
```
If no storage config is supplied, `local` is available and writes under
`storage/app` relative to the process working directory. The local driver is
backed by Flystorage's `FileStorage` and `LocalStorageAdapter`.
## Day-To-Day Usage
Most app code should use the default disk unless it has a good reason to choose
a specific backend. `write(...)` and `put(...)` do the same thing. `put(...)`
matches Laravel naming, while `write(...)` is convenient in the REPL.
### Write A Text File
```ts
await app().storage.write('my/test-file.txt', 'hello i am a test file');
```
Or, using the Laravel-style name:
```ts
await app().storage.put('my/test-file.txt', 'hello i am a test file');
```
### Read A Text File
```ts
const text = await app().storage.readToString('my/test-file.txt');
```
`getText(...)` is also available when you want to pass a Node text encoding.
### Read A Stream
`read(...)` follows Flystorage's API and returns a readable stream:
```ts
const stream = await app().storage.read('my/test-file.txt');
```
Use `get(...)` when you intentionally want the whole file buffered:
```ts
const buffer = await app().storage.readToBuffer('my/test-file.txt');
const bytes = await app().storage.readToUint8Array('my/test-file.txt');
```
`get(...)` is also available as a shorter alias for `readToBuffer(...)`.
### Write JSON
```ts
await app().storage.write('reports/latest.json', JSON.stringify({
status: 'ready',
}, null, '\t'), {
mimeType: 'application/json',
});
```
### Write Bytes
```ts
await app().storage.write('images/avatar.png', imageBytes, {
mimeType: 'image/png',
visibility: 'private',
});
```
### Check, Delete, And Inspect Files
```ts
await app().storage.exists('my/test-file.txt');
await app().storage.missing('my/missing-file.txt');
await app().storage.size('my/test-file.txt');
await app().storage.lastModified('my/test-file.txt');
await app().storage.mimeType('images/avatar.png');
await app().storage.delete('my/test-file.txt');
```
### List Directory Contents
`list(...)` returns a lazy, provider-neutral async listing for local and remote
disks. Set `deep` when descendants below the immediate directory are needed:
```ts
const listing = app().storage.disk('local').list('reports', {
deep: true,
});
for await (const entry of listing) {
console.log(entry.type, entry.path, entry.lastModified);
}
```
Collect a listing when the complete result is intentionally needed in memory:
```ts
const entries = await app().storage.disk('local').list('reports', {
deep: true,
}).toArray();
```
Pass an empty path or omit it to list from the disk root. Listings expose the
same framework entry shape for every driver; callers do not need to know whether
the selected disk is local or S3-compatible.
For local disks, `path(...)` returns the absolute filesystem path:
```ts
app().storage.path('my/test-file.txt');
```
Remote disks such as S3 do not expose local paths, so use `url(...)` when a
public URL is configured:
```ts
await app().storage.url('images/avatar.png');
```
### Use A Named Disk
```ts
await app().storage.disk('agent').write('generated-images/example.txt', 'hello agent disk');
const text = await app().storage.disk('agent').getText('generated-images/example.txt');
```
`drive(name)` is also available as an alias for `disk(name)`:
```ts
await app().storage.drive('agent').write('notes/example.txt', 'stored on the agent disk');
```
## Try It In Tinker
From the repo root, start the CLI REPL:
```sh
npm run tinker
```
An application tinker context can expose both `app` and `app()` styles. Top-level
`await` is supported, so each command waits for the storage operation to finish
before the prompt returns:
```ts
await app().storage.write('my/test-file.txt', 'hello i am a test file')
await app().storage.read('my/test-file.txt')
await app().storage.readToString('my/test-file.txt')
await app().storage.readToBuffer('my/test-file.txt')
await app().storage.readToUint8Array('my/test-file.txt')
await app().storage.exists('my/test-file.txt')
await app().storage.delete('my/test-file.txt')
```
If you prefer property access, this works too:
```ts
await app.storage.write('my/test-file.txt', 'hello from app.storage')
```
## Use The Default Disk
```ts
await app.storage.put('reports/latest.json', JSON.stringify({
status: 'ready',
}));
const text = await app.storage.getText('reports/latest.json');
```
## Use A Named Disk
```ts
const agent = app.storage.disk('agent');
await agent.write('generated-images/example.png', imageBytes, {
mimeType: 'image/png',
visibility: 'private',
});
const bytes = await agent.get('generated-images/example.png');
```
## Stream Large Files
Use explicit stream methods when files should not be buffered into memory:
```ts
await app.storage.disk('agent').writeStream('uploads/source-video.mp4', requestStream, {
mimeType: 'video/mp4',
});
const stream = await app.storage.disk('agent').readStream('uploads/source-video.mp4');
```
`read(...)` is also available and matches Flystorage's stream-returning method:
```ts
const stream = await app.storage.disk('agent').read('uploads/source-video.mp4');
```
These methods are backed by Flystorage streams, so they are the preferred API
for large uploads, exports, and future remote disks such as S3. The framework
wrapper passes stream objects through to the driver and does not buffer them
before writing or reading.
## S3-Compatible Disks
Use `driver: 's3'` for AWS S3 and services that support the S3 protocol:
```ts
const storage = new Storage({
default: 'spaces',
disks: {
spaces: {
driver: 's3',
bucket: 'example-assets',
region: 'nyc3',
endpoint: 'https://nyc3.digitaloceanspaces.com',
url: 'https://example-assets.nyc3.digitaloceanspaces.com',
accessKeyId: env.required('SPACES_ACCESS_KEY_ID'),
secretAccessKey: env.required('SPACES_SECRET_ACCESS_KEY'),
},
},
});
```
`endpoint` is optional for AWS S3 and useful for providers such as DigitalOcean
Spaces, MinIO, and other S3-compatible services. Use `forcePathStyle: true`
when the provider expects path-style bucket URLs.
## Path safety
Storage paths are always relative to the disk root. Absolute paths, null bytes,
and parent-directory traversal are rejected before the driver touches the
filesystem. File operations reject empty paths; `list('')` deliberately allows
an empty path to list the disk root.
## Runnable example and testing
The [storage lab](./examples/runStorage.ts) writes and reads JSON and a streamed
CSV, lists the resulting paths, checks traversal rejection and deletes a file.
It uses a real temporary local disk, removes that directory afterwards and does
not need SQL or cloud credentials. Copy the shipped example into `examples`
and run `npx tsx examples/runStorage.ts` in the consuming app.
The Storage website page includes the exact Vitest test for that copied file.
Framework contributors run `npm run test:service --workspace packages/app -- storage`.
Local behaviour is not evidence that an S3 account, bucket policy or interrupted
upload works correctly. Verify those against dedicated provider test resources.
## Local disk URLs
The local driver can build URLs only when `url` is configured:
```ts
const url = await app.storage.disk('agent').url('generated-images/example.png');
```
This method only returns the URL. Routes remain responsible for authentication,
ownership checks, headers, and response shaping.
````
## Guidance for AI tools
Use the documented public import `@db3.ai/app/storage` and its exported types. Prefer the source-backed examples and behavioural outcomes above over invented APIs or source-relative internal imports.