Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 145 additions & 0 deletions services/kiloclaw/src/durable-objects/kiloclaw-instance.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,8 @@ import { buildChannelConfigPatch } from './kiloclaw-instance/channel-config';
import { destroyRetryDelay } from './kiloclaw-instance/log';
import * as flyClient from '../fly/client';
import { FlyApiError } from '../fly/client';
import type { FlyVolume } from '../fly/types';
import { volumeNameFromSandboxId } from './machine-config';
import * as db from '../db';
import * as gatewayEnv from '../gateway/env';
import * as regions from './regions';
Expand Down Expand Up @@ -435,6 +437,18 @@ async function seedRecovering(
});
}

function flyVolume(overrides: Partial<FlyVolume> & Pick<FlyVolume, 'id'>): FlyVolume {
return {
name: volumeNameFromSandboxId('sandbox-1'),
state: 'created',
size_gb: 10,
region: 'sjc',
attached_machine_id: null,
created_at: '2026-09-14T00:00:00.000Z',
...overrides,
};
}

function dockerProviderState(overrides: Record<string, unknown> = {}) {
return {
provider: 'docker-local',
Expand Down Expand Up @@ -2485,6 +2499,102 @@ describe('reconciliation: volume', () => {
});
expect(dataLossLog).toBeDefined();
});

it('adopts the newest existing volume when flyVolumeId is null after a restart', async () => {
const { instance, storage } = createInstance();
await seedProvisioned(storage, { flyVolumeId: null, lastStartedAt: Date.now() });

const name = volumeNameFromSandboxId('sandbox-1');
(flyClient.listVolumes as Mock).mockResolvedValueOnce([
flyVolume({
id: 'vol-original',
name,
state: 'pending_destroy',
created_at: '2026-09-08T17:39:32.095Z',
}),
flyVolume({ id: 'vol-fork', name, created_at: '2026-09-14T22:21:19.270Z' }),
flyVolume({
id: 'vol-other-sandbox',
name: 'kiloclaw_other',
created_at: '2026-09-16T00:00:00Z',
}),
]);

await instance.alarm();

expect(flyClient.createVolumeWithFallback).not.toHaveBeenCalled();
expect(storage._store.get('flyVolumeId')).toBe('vol-fork');
expect(storage._store.get('flyRegion')).toBe('sjc');
});

it('creates a fresh volume when every matching volume is non-adoptable', async () => {
const { instance, storage } = createInstance();
await seedProvisioned(storage, { flyVolumeId: null, lastStartedAt: Date.now() });

const name = volumeNameFromSandboxId('sandbox-1');
(flyClient.listVolumes as Mock).mockResolvedValueOnce([
flyVolume({ id: 'vol-reaping', name, state: 'pending_destroy' }),
flyVolume({ id: 'vol-gone', name, state: 'destroyed' }),
]);
(flyClient.createVolumeWithFallback as Mock).mockResolvedValue({
id: 'vol-new',
region: 'iad',
});

await instance.alarm();

expect(flyClient.createVolumeWithFallback).toHaveBeenCalled();
expect(storage._store.get('flyVolumeId')).toBe('vol-new');
});

it('does not adopt a matching volume that another machine holds', async () => {
const { instance, storage } = createInstance();
await seedProvisioned(storage, {
flyVolumeId: null,
flyMachineId: null,
lastStartedAt: Date.now(),
});

const name = volumeNameFromSandboxId('sandbox-1');
(flyClient.listVolumes as Mock).mockResolvedValueOnce([
flyVolume({ id: 'vol-orphan', name, attached_machine_id: 'machine-other' }),
]);
(flyClient.createVolumeWithFallback as Mock).mockResolvedValue({
id: 'vol-new',
region: 'iad',
});

await instance.alarm();

expect(storage._store.get('flyVolumeId')).toBe('vol-new');
});

it('does not log data loss when a lost volume pointer is repaired by adopting a fork', async () => {
const { instance, storage } = createInstance();
await seedProvisioned(storage, { flyVolumeId: 'vol-dead', lastStartedAt: Date.now() });

(flyClient.getVolume as Mock).mockRejectedValueOnce(new FlyApiError('not found', 404, '{}'));
(flyClient.listVolumes as Mock).mockResolvedValueOnce([
flyVolume({ id: 'vol-fork', name: volumeNameFromSandboxId('sandbox-1') }),
]);

await instance.alarm();

expect(storage._store.get('flyVolumeId')).toBe('vol-fork');
expect(flyClient.createVolumeWithFallback).not.toHaveBeenCalled();

const replaced = (console.log as Mock).mock.calls
.map((args: unknown[]) => {
try {
return JSON.parse(String(args[0])) as { action?: string; data_loss?: boolean };
} catch {
return null;
}
})
.find(entry => entry?.action === 'replace_lost_volume');
expect(replaced).toBeDefined();
expect(replaced?.data_loss).toBe(false);
});
});

describe('destroying: no recreation', () => {
Expand Down Expand Up @@ -5159,6 +5269,41 @@ describe('start: volume region validation', () => {
// Region was not changed since volume matches stored flyRegion
expect(storage._store.get('flyRegion')).toBe('iad');
});

it('adopts the existing data volume on start when flyVolumeId was lost', async () => {
const { instance, storage } = createInstance();
await seedProvisioned(storage, {
flyVolumeId: null,
flyMachineId: null,
flyRegion: null,
lastStartedAt: Date.now(),
});

(flyClient.listVolumes as Mock).mockResolvedValueOnce([
flyVolume({ id: 'vol-fork', name: volumeNameFromSandboxId('sandbox-1'), region: 'sjc' }),
]);
(flyClient.createMachine as Mock).mockResolvedValue({ id: 'machine-1', region: 'sjc' });
(flyClient.waitForState as Mock).mockResolvedValue(undefined);

await instance.start('user-1');

expect(flyClient.createVolumeWithFallback).not.toHaveBeenCalled();
expect(storage._store.get('flyVolumeId')).toBe('vol-fork');
});

it('start fails closed instead of creating an empty volume when adoption lookup fails', async () => {
const { instance, storage } = createInstance();
await seedProvisioned(storage, {
flyVolumeId: null,
flyMachineId: null,
lastStartedAt: Date.now(),
});

(flyClient.listVolumes as Mock).mockRejectedValueOnce(new FlyApiError('fly down', 500, '{}'));

await expect(instance.start('user-1')).rejects.toThrow('fly down');
expect(flyClient.createVolumeWithFallback).not.toHaveBeenCalled();
});
});

// ============================================================================
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import type { KiloClawEnv } from '../../types';
import type { FlyClientConfig } from '../../fly/client';
import type { FlyMachineConfig } from '../../fly/types';
import type { FlyMachineConfig, FlyVolume } from '../../fly/types';
import type { FlyProviderState } from '../../schemas/instance-config';
import type { RuntimeSpec } from '../../providers/types';
import * as fly from '../../fly/client';
Expand Down Expand Up @@ -73,18 +73,111 @@ export function buildFlyMachineConfig(
};
}

/** Volume states that can never be reused by a new machine. */
const NON_ADOPTABLE_VOLUME_STATES: ReadonlySet<FlyVolume['state']> = new Set([
'pending_destroy',
'destroying',
'destroyed',
]);

/**
* Ensure a Fly Volume exists. Creates one if flyVolumeId is null.
* Adopt the newest existing volume that belongs to this instance.
*
* When the Durable Object loses `flyVolumeId` (for example after an
* unexpected-stop recovery that never finished), creating a fresh volume mounts
* an empty disk over data that still exists on Fly. Recovery forks and
* stranded-volume replacements keep the instance's volume name and are created
* last, so the newest usable volume with that name holds the latest data.
*/
async function adoptExistingVolume(
flyConfig: FlyClientConfig,
expectedName: string,
state: FlyRuntimeState,
reason: string
): Promise<FlyVolume | null> {
const volumes = await fly.listVolumes(flyConfig);
const matching = volumes
.filter(volume => volume.name === expectedName)
.filter(volume => !NON_ADOPTABLE_VOLUME_STATES.has(volume.state))
Comment thread
iscekic marked this conversation as resolved.
.sort((a, b) => Date.parse(b.created_at) - Date.parse(a.created_at));

// A Fly volume is attached to at most one machine. Adopt only volumes that no
// machine holds, or ones the DO's known machine already holds, so a new machine
// never mounts a volume Fly considers in use.
const isReusable = (volume: FlyVolume) =>
volume.attached_machine_id === null || volume.attached_machine_id === state.flyMachineId;
const candidates = matching.filter(isReusable);
const heldElsewhere = matching.filter(volume => !isReusable(volume));

const [adopted] = candidates;
if (!adopted) {
if (heldElsewhere.length > 0) {
doWarn(state, 'Volume adoption skipped: matching volumes are attached to another machine', {
expected_name: expectedName,
volume_ids: heldElsewhere.map(volume => volume.id),
attached_machine_ids: heldElsewhere.map(volume => volume.attached_machine_id),
});
}
return null;
}

if (candidates.length > 1) {
doWarn(state, 'Multiple usable volumes match the instance; adopting the newest', {
expected_name: expectedName,
volume_ids: candidates.map(volume => volume.id),
});
}

reconcileLog(reason, 'adopt_volume', {
fly_app_name: flyConfig.appName,
volume_id: adopted.id,
volume_state: adopted.state,
volume_created_at: adopted.created_at,
region: adopted.region,
});
return adopted;
}

/**
* Ensure a Fly Volume exists.
*
* Reuses the instance's existing volume when the DO lost its reference, so a
* restart cannot mount an empty disk over data that is still on Fly. Creates a
* fresh volume only when no usable volume exists.
*/
export async function ensureVolume(
flyConfig: FlyClientConfig,
state: FlyRuntimeState,
providerState: FlyProviderState,
env: KiloClawEnv,
reason: string
): Promise<FlyProviderState> {
if (providerState.volumeId) return providerState;
if (!state.sandboxId) return providerState;
): Promise<{ providerState: FlyProviderState; adoptedVolumeId: string | null }> {
if (providerState.volumeId) return { providerState, adoptedVolumeId: null };
if (!state.sandboxId) return { providerState, adoptedVolumeId: null };

const expectedName = volumeNameFromSandboxId(state.sandboxId);
let adopted: FlyVolume | null = null;
try {
adopted = await adoptExistingVolume(flyConfig, expectedName, state, reason);
} catch (err) {
// A restart must never fall back to creating an empty volume when the
// lookup fails: that is the data-loss path this adoption exists to stop.
// A never-started instance has no data to protect, so keep old behavior.
if (state.lastStartedAt !== null) throw err;
doWarn(state, 'Volume adoption lookup failed; creating a fresh volume', {
error: toLoggable(err),
});
}
if (adopted) {
Comment thread
iscekic marked this conversation as resolved.
return {
providerState: {
...providerState,
volumeId: adopted.id,
region: adopted.region,
},
adoptedVolumeId: adopted.id,
};
}

const regions = providerState.region
? parseRegions(providerState.region)
Expand All @@ -111,9 +204,12 @@ export async function ensureVolume(
});

return {
...providerState,
volumeId: volume.id,
region: volume.region,
providerState: {
...providerState,
volumeId: volume.id,
region: volume.region,
},
adoptedVolumeId: null,
};
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -771,7 +771,7 @@ async function reconcileVolume(
rctx: ReconcileContext
): Promise<void> {
if (!state.flyVolumeId) {
const providerState = await ensureVolume(
const { providerState } = await ensureVolume(
flyConfig,
state,
getFlyProviderState(state),
Expand Down Expand Up @@ -800,7 +800,7 @@ async function reconcileVolume(
await ctx.storage.put(
storageUpdate(syncProviderStateForStorage(state, { flyVolumeId: null }))
);
const providerState = await ensureVolume(
const { providerState, adoptedVolumeId } = await ensureVolume(
flyConfig,
state,
getFlyProviderState(state),
Expand All @@ -817,7 +817,7 @@ async function reconcileVolume(
)
);
rctx.log('replace_lost_volume', {
data_loss: true,
data_loss: adoptedVolumeId === null,
old_volume_id: oldVolumeId,
new_volume_id: state.flyVolumeId,
durationMs: performance.now() - repairStart,
Expand Down
2 changes: 1 addition & 1 deletion services/kiloclaw/src/providers/fly/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ export const flyProviderAdapter: InstanceProviderAdapter = {

async ensureStorage({ env, state, reason }) {
const flyConfig = getFlyConfig(env, state);
const providerState = await flyMachines.ensureVolume(
const { providerState } = await flyMachines.ensureVolume(
flyConfig,
state,
getFlyProviderState(state),
Expand Down