Infrastructure/store/ — Unified Persistence Layer

Atomic Writes Multi-Backend Sandboxed I/O Crash-Safe

Contents

Architecture

The store follows an atomic write-ahead pattern: validate the target path → write to a PID-stamped temp file → rename into place. All file operations pass through boundary checks, symlink rejection, size limits, and finally the atomic commit.

flowchart TD
  subgraph Callers["Consumer Modules"]
    DEV["device-auth-store.ts"]
    DQ["delivery-queue-storage.ts"]
    EA["exec-approvals-store.ts"]
    SM["state-migrations.ts"]
  end
  subgraph Sync["Sync JSON Backend"]
    JF["json-file.ts\nloadJsonFile / saveJsonFile"]
  end
  subgraph Async["Async Atomic Backend"]
    JFS["json-files.ts\nreadJsonFile / writeJsonAtomic"]
    LOCK["createAsyncLock()\nwithLock()"]
  end
  subgraph SQLite["SQLite Backend"]
    NS["node-sqlite.ts\nrequireNodeSqlite()"]
  end
  subgraph Safety["Safety Layer"]
    FS["fs-safe.ts\nopenFileWithinRoot()"]
    FL["file-lock.ts"]
    FI["file-identity.ts"]
  end
  subgraph Archive["Archive Extraction"]
    AR["archive.ts\nArchiveExtractLimits"]
  end
  Callers --> Sync
  Callers --> Async
  Callers --> SQLite
  Sync --> Safety
  Async --> Safety
  Async --> LOCK
  SQLite --> Safety
  Archive --> Safety
  FS --> FL
  FS --> FI

Data Flow — Atomic Write Lifecycle

This sequence shows how an atomic write flows through the store, from the initial call to the final rename that makes the data visible.

sequenceDiagram
    participant Caller as Consumer Module
    participant Store as writeJsonAtomic()
    participant Lock as AsyncLock
    participant FS as fs-safe.ts
    participant Disk as Filesystem

    Caller->>Store: writeJsonAtomic(filePath, value)
    Store->>Lock: withLock(filePath)
    Lock-->>Store: Lock acquired
    Store->>FS: openFileWithinRoot(dir)
    FS->>FS: Validate path (no traversal)
    FS->>FS: Reject symlinks (O_NOFOLLOW)
    FS-->>Store: SafeOpenResult
    Store->>Disk: write ${filePath}.${pid}.tmp
    Disk-->>Store: Write complete
    Store->>Disk: fsync(fd)
    Disk-->>Store: Synced
    Store->>Disk: rename(tmp, filePath)
    Disk-->>Store: Atomic commit
    Store->>Lock: Release lock
    Lock-->>Caller: Resolved

Delivery Queue Flow

flowchart LR
  ENQ["enqueueDelivery()"] --> DIR["${stateDir}/delivery-queue/"]
  DIR --> LOAD["loadPendingDeliveries()"]
  LOAD --> ATTEMPT["Delivery Attempt"]
  ATTEMPT -->|Success| ACK["ackDelivery()\n(delete file)"]
  ATTEMPT -->|Failure| FAIL["failDelivery()\n(increment retryCount)"]
  FAIL -->|Max retries| MOVE["moveToFailed()\n(archive to failed/)"]
  FAIL -->|Retry| LOAD

Key Classes & Functions

Data Structures

Interactions

Configuration

Directory Map

store/
├── json-file.ts                   ← Synchronous JSON load/save (mode 0o600)
├── json-files.ts                  ← Async atomic JSON: readJsonFile, writeJsonAtomic, createAsyncLock
├── node-sqlite.ts                 ← Dynamic node:sqlite import with warning filter
├── device-auth-store.ts           ← DeviceAuthStore schema, load/store/clear tokens
├── state-migrations.ts            ← Legacy state detection & migration (WhatsApp, agents, pairing)
├── delivery-queue-storage.ts      ← QueuedDelivery: enqueue, load, ack, fail, moveToFailed
├── fs-safe.ts                     ← SafeOpenResult, openFileWithinRoot, readFileWithinRoot
├── archive.ts                     ← ArchiveKind, ArchiveExtractLimits, extraction logic
├── exec-approvals-store.ts        ← Execution approval decision persistence
├── file-lock.ts                   ← File locking primitives for cross-process coordination
├── file-identity.ts               ← Inode/path-based file identity tracking
├── data_models/                   ← Schema definitions and type-safe data model helpers
│   └── ...
├── migrations/                    ← Versioned migration scripts and rollback logic
│   └── ...
└── query_builder/                 ← Composable query construction for SQLite backend
    └── ...

Deep-Dive Highlights

Core persistence mechanisms have been extracted into dedicated focus pages for in-depth exploration. Each page covers one aspect of the store with diagrams, code walks, and configuration details.

Deep-Dive Pages

Code Examples

Atomic JSON Write with Locking

import { writeJsonAtomic, createAsyncLock } from "./json-files";

const lock = createAsyncLock();

async function persistConfig(filePath: string, config: AppConfig): Promise<void> {
  await lock.withLock(async () => {
    await writeJsonAtomic(filePath, config, {
      mode: 0o600,
      mkdir: true,
    });
  });
}

Sandboxed File Read

import { readFileWithinRoot, SafeOpenErrorCode } from "./fs-safe";

async function loadUserFile(root: string, userPath: string): Promise<Buffer | null> {
  const result = await readFileWithinRoot(root, userPath);
  if (!result.ok) {
    switch (result.error) {
      case SafeOpenErrorCode.PATH_TRAVERSAL:
        throw new Error(`Blocked directory traversal: ${userPath}`);
      case SafeOpenErrorCode.SYMLINK:
        throw new Error(`Symlink not allowed: ${userPath}`);
      default:
        return null;
    }
  }
  return result.data;
}

Delivery Queue Lifecycle

import {
  enqueueDelivery,
  loadPendingDeliveries,
  ackDelivery,
  failDelivery,
  moveToFailed,
} from "./delivery-queue-storage";

const delivery: QueuedDelivery = {
  id: crypto.randomUUID(),
  channel: "whatsapp",
  to: "+1234567890",
  payloads: [{ type: "text", body: "Hello from OpenClaw" }],
  enqueuedAt: Date.now(),
  retryCount: 0,
};

await enqueueDelivery(stateDir, delivery);

const pending = await loadPendingDeliveries(stateDir);
for (const item of pending) {
  try {
    await sendMessage(item);
    await ackDelivery(stateDir, item.id);
  } catch (err) {
    await failDelivery(stateDir, item.id, err.message);
    if (item.retryCount >= MAX_RETRIES) {
      await moveToFailed(stateDir, item.id);
    }
  }
}

Device Auth Token Management

import {
  loadDeviceAuthToken,
  storeDeviceAuthToken,
  clearDeviceAuthToken,
} from "./device-auth-store";

// Load existing token (returns undefined if not found)
const auth = await loadDeviceAuthToken(stateDir);

if (!auth || isExpired(auth.tokens)) {
  const freshTokens = await refreshDeviceAuth(auth?.deviceId);
  await storeDeviceAuthToken(stateDir, {
    version: 1,
    deviceId: freshTokens.deviceId,
    tokens: freshTokens.tokens,
  });
}

// On logout, remove persisted credentials
await clearDeviceAuthToken(stateDir);