Skip to content
Closed
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
55 changes: 55 additions & 0 deletions apps/hosted/self-host/src/implementation/durable-declarations.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
/**
* Evaluated declarations and tool listings kept in the self-host's product database, so a restart
* serves the listings it already evaluated instead of loading every app Worker to evaluate them
* again. Cloud keeps the same results in each app's data supervisor.
*
* Results can include text derived from credentials. They stay in the product database, which
* already holds the encrypted account state and is owned by the same operator.
*/
import { Effect } from "effect";
import { SqlClient } from "effect/sql";
import type { DeclarationCache, DurableDeclarations } from "@executor-js/sdk/core";

/** Results larger than this are not kept; a cold evaluation is cheaper than storing them. */
const entryChars = 4 * 1024 * 1024;

export const selfHostDurableDeclarations = Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
const durable: DurableDeclarations = {
get: (app, key) =>
sql<{ at: string | number; json: string }>`select at, json from hosted_evaluated
where app = ${app} and key = ${key} and until > ${Date.now()}`.pipe(
Effect.map((rows) => {
const row = rows[0];
return row === undefined ? undefined : { at: Number(row.at), json: row.json };
}),
Effect.catchCause(() => Effect.succeed(undefined)),
Effect.withSpan("storage.evaluated.read"),
),
set: (app, key, entry) =>
entry.json.length > entryChars
? Effect.void
: sql`insert into hosted_evaluated (app, key, at, until, json)
values (${app}, ${key}, ${entry.at}, ${entry.until}, ${entry.json})
on conflict (app, key) do update
set at = excluded.at, until = excluded.until, json = excluded.json
where hosted_evaluated.at <= excluded.at`.pipe(
Effect.asVoid,
Effect.catchCause(() => Effect.logWarning("Evaluated result was not kept")),
Effect.withSpan("storage.evaluated.write"),
),
};
/** The app's durable results are forgotten with its in-process ones when its cache changes. */
const forgetting = (cache: DeclarationCache): DeclarationCache => ({
...cache,
changed: (app, at) => {
cache.changed(app, at);
Effect.runFork(
sql`delete from hosted_evaluated where app = ${app} and at <= ${at}`.pipe(
Effect.catchCause(() => Effect.logWarning("Evaluated results were not forgotten")),
),
);
},
});
return { durable, forgetting };
});
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ import type { AppRuntime, BlobStorage, WorkflowRuntime } from "@executor-js/sdk/
import { Config, Effect, Layer, Option, Deferred, Schedule, Context, Scope } from "effect";
import { GroupDatabase } from "@executor-js/hosted-server/groups";
import { SqlClient } from "effect/sql";
import { selfHostDurableDeclarations } from "./durable-declarations.ts";

/** Native resources supplied at the self-host composition boundary. */
export interface SelfHostPlatform {
Expand Down Expand Up @@ -79,6 +80,7 @@ export const selfHostExecutorServices = <E, R>(
Config.withDefault(hostedExecutorOrigin),
),
);
const kept = yield* selfHostDurableDeclarations;
const executor = yield* createExecutor({
database: storage,
secret: key,
Expand All @@ -96,7 +98,8 @@ export const selfHostExecutorServices = <E, R>(
...(Option.isSome(clientMetadata) ? { clientMetadataUrl: clientMetadata.value.url } : {}),
},
cache: {
memory: makeDeclarationCache(evaluation.limits),
memory: kept.forgetting(makeDeclarationCache(evaluation.limits)),
durable: kept.durable,
toolListings: evaluation.toolListings,
},
// Stale declarations refresh on the server's own lifetime.
Expand Down
15 changes: 15 additions & 0 deletions apps/hosted/server/src/implementation/evaluated-schema.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
import { Effect } from "effect";
import { SqlClient } from "effect/sql";

/** Evaluated declarations and tool listings a self-hosted server keeps across restarts. */
export const migrateEvaluatedDeclarations = Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
yield* sql`create table hosted_evaluated (
app text not null,
key text not null,
at bigint not null,
until bigint not null,
json text not null,
primary key (app, key)
)`;
});
3 changes: 3 additions & 0 deletions apps/hosted/server/src/implementation/migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import { migrateResourceAccess } from "./resource-schema.ts";
import { migrateOrganizationRemovals } from "./organization-removal-schema.ts";
import { migrateApiKeyMemberships } from "./api-key-membership-schema.ts";
import { queueExecutorAppUpgrades } from "./executor-app-upgrades.ts";
import { migrateEvaluatedDeclarations } from "./evaluated-schema.ts";

/** Migration failures stop startup; callers must not log the driver's secret-bearing cause. */
export class HostedMigrationFailed extends Schema.TaggedError<HostedMigrationFailed>()(
Expand Down Expand Up @@ -63,6 +64,8 @@ const hostedProductMigrations = migrateProductSteps("private_hosted_migrations",
"5_upgrade_executor_apps": queueExecutorAppUpgrades,
// Step 5 skipped apps whose only change since deployment was the framework pin commit.
"6_upgrade_pinned_executor_apps": queueExecutorAppUpgrades,
// Additive: the running server never reads it.
"7_evaluated_declarations": migrateEvaluatedDeclarations,
});

/**
Expand Down