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
5 changes: 4 additions & 1 deletion apps/desktop/src/app/DesktopAppActivation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -382,7 +382,10 @@ export const make = Effect.gen(function* () {
webContents.send(DESKTOP_APP_ACTIVATION_REQUEST_CHANNEL, request);
});
}),
complete: (response) => Effect.sync(() => broker.complete(response)),
// The IPC handler owns the activation-completion span.
complete: Effect.fnUntraced(function* (response) {
return yield* Effect.sync(() => broker.complete(response));
}),
});
});

Expand Down
169 changes: 86 additions & 83 deletions apps/desktop/src/app/DesktopObservability.ts
Original file line number Diff line number Diff line change
Expand Up @@ -459,90 +459,91 @@ const makeBackendOutputLogShape = (
): Effect.Effect<DesktopBackendOutputLogShape> =>
Option.match(sink, {
onNone: () => Effect.succeed(DesktopBackendOutputLogNoop),
onSome: (logFile) =>
Effect.gen(function* () {
const sessionRef = yield* Ref.make(Option.none<BackendOutputSession>());
const writeFailure = Effect.fn("desktop.observability.backendOutput.writeFailure")(
function* (session: BackendOutputSession, details: string) {
yield* writeBackendChildLogRecord(logFile, {
message: "backend child process failure output start",
level: "ERROR",
annotations: {
component: "desktop-backend-child",
runId: session.runId,
instanceId: id,
phase: "START",
details: session.startDetails,
},
});
for (const output of session.chunks) {
yield* writeBackendChildLogRecord(logFile, {
message: "backend child process output",
level: output.streamName === "stderr" ? "ERROR" : "INFO",
annotations: {
component: "desktop-backend-child",
runId: session.runId,
instanceId: id,
stream: output.streamName,
text: textDecoder.decode(output.chunk.subarray(output.offset)),
},
});
}
yield* writeBackendChildLogRecord(logFile, {
message: "backend child process failure output end",
level: "ERROR",
annotations: {
component: "desktop-backend-child",
runId: session.runId,
instanceId: id,
phase: "END",
details: sanitizeLogValue(details),
},
});
// The enclosing operation owns the span for this inline handler.
onSome: Effect.fnUntraced(function* (logFile) {
const sessionRef = yield* Ref.make(Option.none<BackendOutputSession>());
const writeFailure = Effect.fn("desktop.observability.backendOutput.writeFailure")(function* (
session: BackendOutputSession,
details: string,
) {
yield* writeBackendChildLogRecord(logFile, {
message: "backend child process failure output start",
level: "ERROR",
annotations: {
component: "desktop-backend-child",
runId: session.runId,
instanceId: id,
phase: "START",
details: session.startDetails,
},
);
return {
beginSession: Effect.fn("desktop.observability.backendOutput.beginSession")(function* ({
details,
}) {
const runId = yield* currentDesktopRunId;
yield* Ref.set(
sessionRef,
Option.some({
runId,
startDetails: sanitizeLogValue(details),
chunks: [],
byteLength: 0,
}),
);
}),
writeOutputChunk: Effect.fnUntraced(function* (streamName, chunk) {
if (environment.isDevelopment) {
yield* writeDevelopmentConsoleOutput(streamName, chunk);
}
yield* Ref.update(
sessionRef,
Option.map((session) => appendBoundedOutputChunk(session, streamName, chunk)),
);
}),
persistFailureSnapshot: Effect.fn(
"desktop.observability.backendOutput.persistFailureSnapshot",
)(function* ({ details }) {
const session = yield* Ref.get(sessionRef);
if (Option.isSome(session)) {
yield* writeFailure(session.value, details);
}
}),
persistFailure: Effect.fn("desktop.observability.backendOutput.persistFailure")(
function* ({ details }) {
const session = yield* Ref.modify(sessionRef, (current) => [current, Option.none()]);
if (Option.isNone(session)) return;
yield* writeFailure(session.value, details);
});
for (const output of session.chunks) {
yield* writeBackendChildLogRecord(logFile, {
message: "backend child process output",
level: output.streamName === "stderr" ? "ERROR" : "INFO",
annotations: {
component: "desktop-backend-child",
runId: session.runId,
instanceId: id,
stream: output.streamName,
text: textDecoder.decode(output.chunk.subarray(output.offset)),
},
),
discardSession: Ref.set(sessionRef, Option.none()),
} satisfies DesktopBackendOutputLogShape;
}),
});
}
yield* writeBackendChildLogRecord(logFile, {
message: "backend child process failure output end",
level: "ERROR",
annotations: {
component: "desktop-backend-child",
runId: session.runId,
instanceId: id,
phase: "END",
details: sanitizeLogValue(details),
},
});
});
return {
beginSession: Effect.fn("desktop.observability.backendOutput.beginSession")(function* ({
details,
}) {
const runId = yield* currentDesktopRunId;
yield* Ref.set(
sessionRef,
Option.some({
runId,
startDetails: sanitizeLogValue(details),
chunks: [],
byteLength: 0,
}),
);
}),
writeOutputChunk: Effect.fnUntraced(function* (streamName, chunk) {
if (environment.isDevelopment) {
yield* writeDevelopmentConsoleOutput(streamName, chunk);
}
yield* Ref.update(
sessionRef,
Option.map((session) => appendBoundedOutputChunk(session, streamName, chunk)),
);
}),
persistFailureSnapshot: Effect.fn(
"desktop.observability.backendOutput.persistFailureSnapshot",
)(function* ({ details }) {
const session = yield* Ref.get(sessionRef);
if (Option.isSome(session)) {
yield* writeFailure(session.value, details);
}
}),
persistFailure: Effect.fn("desktop.observability.backendOutput.persistFailure")(function* ({
details,
}) {
const session = yield* Ref.modify(sessionRef, (current) => [current, Option.none()]);
if (Option.isNone(session)) return;
yield* writeFailure(session.value, details);
}),
discardSession: Ref.set(sessionRef, Option.none()),
} satisfies DesktopBackendOutputLogShape;
}),
});

const layerBackendOutputLogFactory = Layer.effect(
Expand Down Expand Up @@ -589,7 +590,9 @@ const layerBackendOutputLogFactory = Layer.effect(
});

return DesktopBackendOutputLogFactory.of({
forInstance: (id) => makeForId(id),
forInstance: Effect.fn("DesktopBackendOutputLogFactory.forInstance")(function (id) {
return makeForId(id);
}),
});
}),
);
Expand Down
15 changes: 7 additions & 8 deletions apps/desktop/src/app/DesktopWebLinks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -69,16 +69,15 @@ const make = Effect.gen(function* () {
});

return DesktopWebLinks.of({
receive: (url) =>
Effect.suspend(() => {
receive: Effect.fn("DesktopWebLinks.receive")(function (url) {
return Effect.suspend(() => {
pending.push(url);
return flush;
}).pipe(Effect.withSpan("DesktopWebLinks.receive")),
setRendererReady: (ready) =>
(ready ? startListening : Effect.sync(stopListening)).pipe(
Effect.andThen(flush),
Effect.withSpan("DesktopWebLinks.setRendererReady"),
),
});
}),
setRendererReady: Effect.fn("DesktopWebLinks.setRendererReady")(function (ready) {
return ready ? startListening : Effect.sync(stopListening);
}, Effect.andThen(flush)),
});
});

Expand Down
96 changes: 48 additions & 48 deletions apps/desktop/src/backend/DesktopBackendManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1043,58 +1043,58 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
});
yield* Option.match(active, {
onNone: () => Effect.void,
onSome: (run) =>
Effect.gen(function* () {
const closed = yield* closeRun(run, parentScope, options);
if (!closed) {
return;
}
const cleanup = yield* mutex.withPermits(1)(
Ref.modify(
state,
(
latest,
): readonly [
{
readonly needsCleanup: boolean;
readonly shouldStart: boolean;
},
BackendManagerState,
] => {
const current = Option.getOrUndefined(latest.active);
if (current?.id !== run.id) {
return [
{
needsCleanup: false,
shouldStart:
latest.desiredRunning &&
Option.isNone(latest.active) &&
Option.isNone(latest.restartFiber),
},
latest,
];
}
// The enclosing operation owns the span for this inline handler.
onSome: Effect.fnUntraced(function* (run) {
const closed = yield* closeRun(run, parentScope, options);
if (!closed) {
return;
}
const cleanup = yield* mutex.withPermits(1)(
Ref.modify(
state,
(
latest,
): readonly [
{
readonly needsCleanup: boolean;
readonly shouldStart: boolean;
},
BackendManagerState,
] => {
const current = Option.getOrUndefined(latest.active);
if (current?.id !== run.id) {
return [
{
needsCleanup: true,
shouldStart: latest.desiredRunning,
},
{
...latest,
active: Option.none<ActiveBackendRun>(),
needsCleanup: false,
shouldStart:
latest.desiredRunning &&
Option.isNone(latest.active) &&
Option.isNone(latest.restartFiber),
},
latest,
];
},
),
);
if (cleanup.needsCleanup) {
yield* desktopTelemetryPublisher.removeControlSource(spec.id);
yield* backendOutputLog.discardSession;
}
if (cleanup.shouldStart) {
yield* start;
}
}),
}
return [
{
needsCleanup: true,
shouldStart: latest.desiredRunning,
},
{
...latest,
active: Option.none<ActiveBackendRun>(),
},
];
},
),
);
if (cleanup.needsCleanup) {
yield* desktopTelemetryPublisher.removeControlSource(spec.id);
yield* backendOutputLog.discardSession;
}
if (cleanup.shouldStart) {
yield* start;
}
}),
});
});

Expand Down
19 changes: 13 additions & 6 deletions apps/desktop/src/backend/DesktopBackendPool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -291,13 +291,14 @@ export const layer = Layer.effect(
});

return DesktopBackendPool.of({
get: (id) =>
SynchronizedRef.get(instancesRef).pipe(
get: Effect.fn("DesktopBackendPool.get")(function (id) {
return SynchronizedRef.get(instancesRef).pipe(
Effect.map((instances) => {
const entry = instances.get(id);
return entry?._tag === "Active" ? Option.some(entry.instance) : Option.none();
}),
),
);
}),
list: SynchronizedRef.get(instancesRef).pipe(
Effect.map((instances) =>
Array.from(instances.values()).flatMap((entry) =>
Expand Down Expand Up @@ -326,11 +327,17 @@ export const layerTest = (
);
const primary = instances[0]!;
return DesktopBackendPool.of({
get: (id) => Effect.succeed(Option.fromNullishOr(byId.get(id))),
get: Effect.fn("DesktopBackendPool.get")(function (id) {
return Effect.succeed(Option.fromNullishOr(byId.get(id)));
}),
list: Effect.succeed(Array.from(byId.values())),
primary: Effect.succeed(primary),
register: () => Effect.die("DesktopBackendPool.layerTest does not support register"),
unregister: () => Effect.die("DesktopBackendPool.layerTest does not support unregister"),
register: Effect.fn("DesktopBackendPool.register")(function () {
return Effect.die("DesktopBackendPool.layerTest does not support register");
}),
unregister: Effect.fn("DesktopBackendPool.unregister")(function () {
return Effect.die("DesktopBackendPool.layerTest does not support unregister");
}),
});
}),
);
Loading
Loading