atched issue with duration

Author: matthew-sessions

packages/task/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@durable-effect/task",
-  "version": "0.0.4",
+  "version": "0.0.7",
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -28,7 +28,7 @@
     "test:watch": "vitest"
   },
   "peerDependencies": {
-    "effect": "^4.0.0-beta.21",
+    "effect": "^4.0.0-beta.38",
     "@cloudflare/workers-types": ">=4.0.0"
   },
   "peerDependenciesMeta": {
@@ -37,7 +37,7 @@
     }
   },
   "devDependencies": {
-    "effect": "4.0.0-beta.21",
+    "effect": "4.0.0-beta.38",
     "vitest": "^2.1.0"
   }
 }

packages/task/src/Task.ts

@@ -7,7 +7,7 @@ import type { TaskDefineConfig, TaskDefinition } from "./TaskDefinition.js"
 export const Task = {
   define<S, E, EErr, AErr, R, OErr = never, GErr = never>(
     config: TaskDefineConfig<S, E, EErr, AErr, R, OErr, GErr>,
-  ): TaskDefinition<S, E, EErr | AErr | OErr | GErr, R> {
+  ): TaskDefinition<S, E, EErr, AErr, R, OErr, GErr> {
     return {
       _tag: "TaskDefinition",
       state: config.state,

packages/task/src/TaskDefinition.ts

@@ -38,7 +38,7 @@ export interface TaskDefineConfig<
   ) => Effect.Effect<void, AErr, R>
   readonly onError?: (
     ctx: TaskContext<S>,
-    error: unknown,
+    error: EErr | AErr,
   ) => Effect.Effect<void, OErr, R>
   readonly onClientGetState?: (
     ctx: TaskContext<S>,
@@ -66,44 +66,53 @@ export interface TaskDefineConfigVoid<
   ) => Effect.Effect<void, AErr, R>
   readonly onError?: (
     ctx: TaskContext<S>,
-    error: unknown,
+    error: EErr | AErr,
   ) => Effect.Effect<void, OErr, R>
 }
 
 // ---------------------------------------------------------------------------
 // TaskDefinition — the pure definition value
 // ---------------------------------------------------------------------------
 
-export interface TaskDefinition<S, E, Err, R> {
+export interface TaskDefinition<S, E, EErr, AErr, R, OErr = never, GErr = never> {
   readonly _tag: "TaskDefinition"
   readonly state: PureSchema<S>
   readonly event: PureSchema<E>
   readonly onEvent: (
     ctx: TaskContext<S>,
     event: E,
-  ) => Effect.Effect<void, Err, R>
+  ) => Effect.Effect<void, EErr, R>
   readonly onAlarm: (
     ctx: TaskContext<S>,
-  ) => Effect.Effect<void, Err, R>
+  ) => Effect.Effect<void, AErr, R>
   readonly onError?: (
     ctx: TaskContext<S>,
-    error: unknown,
-  ) => Effect.Effect<void, Err, R>
+    error: EErr | AErr,
+  ) => Effect.Effect<void, OErr, R>
   readonly onClientGetState?: (
     ctx: TaskContext<S>,
     state: S | null,
-  ) => Effect.Effect<S | null, Err, R>
+  ) => Effect.Effect<S | null, GErr, R>
 }
 
+// ---------------------------------------------------------------------------
+// TaskErrors — extract the full error union from a TaskDefinition
+// ---------------------------------------------------------------------------
+
+export type TaskErrors<D> =
+  D extends TaskDefinition<any, any, infer EErr, infer AErr, any, infer OErr, infer GErr>
+    ? EErr | AErr | OErr | GErr
+    : never
+
 // ---------------------------------------------------------------------------
 // withServices — wraps handlers with Effect.provide to eliminate R,
 // returning TaskDefinition<S, E, Err, never> (preserving S and E).
 // ---------------------------------------------------------------------------
 
-export function withServices<S, E, Err, R>(
-  definition: TaskDefinition<S, E, Err, R>,
+export function withServices<S, E, EErr, AErr, R, OErr, GErr>(
+  definition: TaskDefinition<S, E, EErr, AErr, R, OErr, GErr>,
   layer: Layer.Layer<R>,
-): TaskDefinition<S, E, Err, never> {
+): TaskDefinition<S, E, EErr, AErr, never, OErr, GErr> {
   return {
     _tag: "TaskDefinition",
     state: definition.state,

packages/task/src/cloudflare/createTasks.ts

@@ -11,14 +11,14 @@ import { TaskClientError } from "./errors.js"
 // ---------------------------------------------------------------------------
 
 export type EventOf<
-  T extends Record<string, TaskDefinition<any, any, any, never>>,
+  T extends Record<string, TaskDefinition<any, any, any, any, never>>,
   K extends keyof T,
-> = T[K] extends TaskDefinition<any, infer E, any, any> ? E : never
+> = T[K] extends TaskDefinition<any, infer E, any, any, any, any, any> ? E : never
 
 export type StateOf<
-  T extends Record<string, TaskDefinition<any, any, any, never>>,
+  T extends Record<string, TaskDefinition<any, any, any, any, never>>,
   K extends keyof T,
-> = T[K] extends TaskDefinition<infer S, any, any, any> ? S : never
+> = T[K] extends TaskDefinition<infer S, any, any, any, any, any, any> ? S : never
 
 // ---------------------------------------------------------------------------
 // Structural types — avoids depending on CF ambient types
@@ -46,7 +46,7 @@ export interface TaskHandle<S, E> {
 // ---------------------------------------------------------------------------
 
 export interface TasksAccessor<
-  T extends Record<string, TaskDefinition<any, any, any, never>>,
+  T extends Record<string, TaskDefinition<any, any, any, any, never>>,
 > {
   <K extends keyof T & string>(
     doNamespace: DurableObjectNamespaceLike,
@@ -59,7 +59,7 @@ export interface TasksAccessor<
 // ---------------------------------------------------------------------------
 
 export function createTasks<
-  const T extends Record<string, TaskDefinition<any, any, any, never>>,
+  const T extends Record<string, TaskDefinition<any, any, any, any, never>>,
 >(
   definitions: T,
 ): {

packages/task/src/index.ts

@@ -14,7 +14,7 @@ export { Alarm, AlarmError } from "./services/Alarm.js"
 // User-facing API
 export { Task } from "./Task.js"
 export type { TaskContext } from "./TaskContext.js"
-export type { PureSchema, TaskDefineConfig, TaskDefinition } from "./TaskDefinition.js"
+export type { PureSchema, TaskDefineConfig, TaskDefinition, TaskErrors } from "./TaskDefinition.js"
 export { withServices } from "./TaskDefinition.js"
 
 // Framework services

packages/task/src/services/TaskRegistry.ts

@@ -95,7 +95,7 @@ function buildTaskContext<S>(
 
   const scheduleIn = (delay: Duration.Input): Effect.Effect<void, TaskError> =>
     Effect.gen(function* () {
-      const ms = Duration.toMillis(delay)
+      const ms = Duration.toMillis(Duration.fromInputUnsafe(delay))
       const ts = Date.now() + ms
       yield* alarm.set(ts).pipe(Effect.mapError(mapAlarmError))
       yield* storage.set(ALARM_KEY, ts).pipe(Effect.mapError(mapStorageError))
@@ -140,18 +140,53 @@ function cleanup(
   }).pipe(Effect.mapError((e) => new TaskExecutionError({ cause: e })))
 }
 
+// ---------------------------------------------------------------------------
+// Internal: type guard for PurgeSignal
+// ---------------------------------------------------------------------------
+
+function isPurgeSignal(error: unknown): error is PurgeSignal {
+  return error instanceof PurgeSignal
+}
+
+// ---------------------------------------------------------------------------
+// Internal: handle errors from a handler effect, routing PurgeSignal to
+// cleanup and other errors to onError (if provided).
+// ---------------------------------------------------------------------------
+
+function handleHandlerError<S, HErr, OErr>(
+  ctx: TaskContext<S>,
+  error: HErr,
+  onError: ((ctx: TaskContext<S>, error: HErr) => Effect.Effect<void, OErr, never>) | undefined,
+  storage: Storage["Service"],
+  alarm: Alarm["Service"],
+): Effect.Effect<void, TaskExecutionError> {
+  if (isPurgeSignal(error)) {
+    return cleanup(storage, alarm)
+  }
+
+  if (!onError) {
+    return Effect.fail(new TaskExecutionError({ cause: error }))
+  }
+
+  return onError(ctx, error).pipe(
+    Effect.catch((oErr) =>
+      isPurgeSignal(oErr)
+        ? cleanup(storage, alarm)
+        : Effect.fail(new TaskExecutionError({ cause: oErr })),
+    ),
+    Effect.mapError((e) => new TaskExecutionError({ cause: e })),
+  )
+}
+
 // ---------------------------------------------------------------------------
 // Internal: build RegisteredTask from a fully-resolved definition (R = never).
 //
 // All handler effects in the definition have R = never, so no layer provision
-// is needed. S, E, Err are captured in the closures via the generic params.
-//
-// The error widening trick (mapError to Err | PurgeSignal) lets catchTag
-// find the PurgeSignal tag even when Err is a generic type parameter.
+// is needed. S, E, EErr, AErr are captured in closures via generic params.
 // ---------------------------------------------------------------------------
 
-function buildRegisteredTask<S, E, Err>(
-  definition: TaskDefinition<S, E, Err, never>,
+function buildRegisteredTask<S, E, EErr, AErr, OErr, GErr>(
+  definition: TaskDefinition<S, E, EErr, AErr, never, OErr, GErr>,
 ): RegisteredTask {
   const decodeEvent = Schema.decodeUnknownEffect(definition.event)
   const decodeState = Schema.decodeUnknownEffect(definition.state)
@@ -171,26 +206,9 @@ function buildRegisteredTask<S, E, Err>(
 
       const ctx = buildTaskContext(storage, alarm, id, name, decodeState, encodeState)
 
-      // Widen error to include PurgeSignal so catchTag can resolve the tag
-      const withPurge = definition.onEvent(ctx, event).pipe(
-        Effect.mapError((e): Err | PurgeSignal => e),
-        Effect.catchTag("PurgeSignal", () => cleanup(storage, alarm)),
-      )
-
-      if (!definition.onError) {
-        yield* withPurge.pipe(
-          Effect.mapError((e) => new TaskExecutionError({ cause: e })),
-        )
-        return
-      }
-
-      yield* withPurge.pipe(
+      yield* definition.onEvent(ctx, event).pipe(
         Effect.catch((error) =>
-          definition.onError!(ctx, error).pipe(
-            Effect.mapError((e): Err | PurgeSignal => e),
-            Effect.catchTag("PurgeSignal", () => cleanup(storage, alarm)),
-            Effect.mapError((e) => new TaskExecutionError({ cause: e })),
-          )
+          handleHandlerError(ctx, error, definition.onError, storage, alarm),
         ),
       )
     })
@@ -202,27 +220,16 @@ function buildRegisteredTask<S, E, Err>(
     name: string,
   ): Effect.Effect<void, TaskExecutionError> =>
     Effect.gen(function* () {
-      const ctx = buildTaskContext(storage, alarm, id, name, decodeState, encodeState)
-
-      const withPurge = definition.onAlarm(ctx).pipe(
-        Effect.mapError((e): Err | PurgeSignal => e),
-        Effect.catchTag("PurgeSignal", () => cleanup(storage, alarm)),
+      // Clear the alarm bookmark — the alarm has fired, so it's no longer pending
+      yield* storage.delete(ALARM_KEY).pipe(
+        Effect.mapError((e) => new TaskExecutionError({ cause: e })),
       )
 
-      if (!definition.onError) {
-        yield* withPurge.pipe(
-          Effect.mapError((e) => new TaskExecutionError({ cause: e })),
-        )
-        return
-      }
+      const ctx = buildTaskContext(storage, alarm, id, name, decodeState, encodeState)
 
-      yield* withPurge.pipe(
+      yield* definition.onAlarm(ctx).pipe(
         Effect.catch((error) =>
-          definition.onError!(ctx, error).pipe(
-            Effect.mapError((e): Err | PurgeSignal => e),
-            Effect.catchTag("PurgeSignal", () => cleanup(storage, alarm)),
-            Effect.mapError((e) => new TaskExecutionError({ cause: e })),
-          )
+          handleHandlerError(ctx, error, definition.onError, storage, alarm),
         ),
       )
     })
@@ -258,8 +265,8 @@ function buildRegisteredTask<S, E, Err>(
 // registerTask — for definitions with no service requirements (R = never)
 // ---------------------------------------------------------------------------
 
-export function registerTask<S, E, Err>(
-  definition: TaskDefinition<S, E, Err, never>,
+export function registerTask<S, E, EErr, AErr, OErr, GErr>(
+  definition: TaskDefinition<S, E, EErr, AErr, never, OErr, GErr>,
 ): RegisteredTask {
   return buildRegisteredTask(definition)
 }
@@ -272,11 +279,11 @@ export function registerTask<S, E, Err>(
 // shared buildRegisteredTask.
 // ---------------------------------------------------------------------------
 
-export function registerTaskWithLayer<S, E, Err, R>(
-  definition: TaskDefinition<S, E, Err, R>,
+export function registerTaskWithLayer<S, E, EErr, AErr, R, OErr, GErr>(
+  definition: TaskDefinition<S, E, EErr, AErr, R, OErr, GErr>,
   layer: Layer.Layer<R>,
 ): RegisteredTask {
-  const resolved: TaskDefinition<S, E, Err, never> = {
+  const resolved: TaskDefinition<S, E, EErr, AErr, never, OErr, GErr> = {
     _tag: "TaskDefinition",
     state: definition.state,
     event: definition.event,