backport effect v4 cluster

Author: tim-smart

.changeset/loud-cows-prove.md

@@ -0,0 +1,34 @@
+---
+"@effect/platform-node-shared": minor
+"@effect/platform-node": minor
+"@effect/platform-bun": minor
+"@effect/cluster": minor
+"@effect/rpc": patch
+"@effect/workflow": minor
+---
+
+backport @effect/cluster from effect v4
+
+@effect/cluster no longer requires a Shard Manager, and instead relies on the
+`RunnerStorage` service to track runner state.
+
+To migrate, remove any Shard Manager deployments and use the updated layers in
+`@effect/platform-node` or `@effect/platform-bun`.
+
+# Breaking Changes
+
+- `ShardManager` module has been removed
+- `@effect/platform-node/NodeClusterSocketRunner` is now
+  `@effect/cluster/NodeClusterSocket`
+- `@effect/platform-node/NodeClusterHttpRunner` is now
+  `@effect/cluster/NodeClusterHttp`
+- `@effect/platform-bun/BunClusterSocketRunner` is now
+  `@effect/cluster/BunClusterSocket`
+- `@effect/platform-bun/BunClusterHttpRunner` is now
+  `@effect/cluster/BunClusterHttp`
+
+# New Features
+
+- `RunnerHealth.layerK8s` has been added, which uses the Kubernetes API to track
+  runner health and liveness. To use it, you will need a service account with
+  permissions to read pod information.

.changeset/petite-signs-thank.md

@@ -0,0 +1,5 @@
+---
+"@effect/sql-pg": patch
+---
+
+disable pg onnotice by default

.changeset/plenty-bats-ask.md

@@ -0,0 +1,5 @@
+---
+"@effect/platform": patch
+---
+
+expose Layer output in HttpLayerRouter.serve

.changeset/sad-bags-fall.md

@@ -0,0 +1,5 @@
+---
+"@effect/sql-pg": minor
+---
+
+Use "pg" npm library for @effect/sql-pg backend

.changeset/warm-aliens-dig.md

@@ -0,0 +1,5 @@
+---
+"effect": minor
+---
+
+add experimental HashRing module

packages/cluster/src/ClusterMetrics.ts

@@ -3,12 +3,6 @@
  */
 import * as Metric from "effect/Metric"
 
-/**
- * @since 1.0.0
- * @category metrics
- */
-export const shards = Metric.gauge("effect_cluster_shards")
-
 /**
  * @since 1.0.0
  * @category metrics
@@ -21,28 +15,30 @@ export const entities = Metric.gauge("effect_cluster_entities", {
  * @since 1.0.0
  * @category metrics
  */
-export const singletons = Metric.gauge("effect_cluster_singletons")
-
-/**
- * @since 1.0.0
- * @category metrics
- */
-export const runners = Metric.gauge("effect_cluster_runners")
+export const singletons = Metric.gauge("effect_cluster_singletons", {
+  bigint: true
+})
 
 /**
  * @since 1.0.0
  * @category metrics
  */
-export const assignedShards = Metric.gauge("effect_cluster_shards_assigned")
+export const runners = Metric.gauge("effect_cluster_runners", {
+  bigint: true
+})
 
 /**
  * @since 1.0.0
  * @category metrics
  */
-export const unassignedShards = Metric.gauge("effect_cluster_shards_unassigned")
+export const runnersHealthy = Metric.gauge("effect_cluster_runners_healthy", {
+  bigint: true
+})
 
 /**
  * @since 1.0.0
  * @category metrics
  */
-export const rebalances = Metric.counter("effect_cluster_rebalances")
+export const shards = Metric.gauge("effect_cluster_shards", {
+  bigint: true
+})

packages/cluster/src/ClusterWorkflowEngine.ts

@@ -2,24 +2,27 @@
  * @since 1.0.0
  */
 import * as Rpc from "@effect/rpc/Rpc"
+import * as RpcServer from "@effect/rpc/RpcServer"
 import { DurableDeferred } from "@effect/workflow"
 import * as Activity from "@effect/workflow/Activity"
 import * as DurableClock from "@effect/workflow/DurableClock"
 import * as Workflow from "@effect/workflow/Workflow"
 import { WorkflowEngine, WorkflowInstance } from "@effect/workflow/WorkflowEngine"
 import * as Arr from "effect/Array"
+import * as Cause from "effect/Cause"
 import * as Context from "effect/Context"
 import * as DateTime from "effect/DateTime"
 import * as Duration from "effect/Duration"
 import * as Effect from "effect/Effect"
 import type * as Exit from "effect/Exit"
 import * as Fiber from "effect/Fiber"
+import * as FiberId from "effect/FiberId"
 import * as Layer from "effect/Layer"
 import * as Option from "effect/Option"
 import type * as ParseResult from "effect/ParseResult"
 import * as PrimaryKey from "effect/PrimaryKey"
 import * as RcMap from "effect/RcMap"
-import * as Record from "effect/Record"
+import type * as Record from "effect/Record"
 import * as Runtime from "effect/Runtime"
 import * as Schedule from "effect/Schedule"
 import * as Schema from "effect/Schema"
@@ -260,13 +263,12 @@ export const make = Effect.gen(function*() {
             return {
               run: (request: Entity.Request<any>) => {
                 const instance = WorkflowInstance.initial(workflow, executionId)
-                let payload = request.payload
+                const payload = request.payload
                 let parent: { workflowName: string; executionId: string } | undefined
                 if (payload[payloadParentKey]) {
                   parent = payload[payloadParentKey]
-                  payload = Record.remove(payload, payloadParentKey)
                 }
-                return execute(payload, executionId).pipe(
+                return execute(workflow.payloadSchema.make(payload), executionId).pipe(
                   Effect.ensuring(Effect.suspend(() => {
                     if (!instance.suspended) {
                       return parent ? ensureSuccess(sendResumeParent(parent)) : Effect.void
@@ -291,17 +293,17 @@ export const make = Effect.gen(function*() {
                 ) as any
               },
 
-              activity: Effect.fnUntraced(
-                function*(request: Entity.Request<any>) {
-                  const activityId = `${executionId}/${request.payload.name}`
+              activity(request: Entity.Request<any>) {
+                const activityId = `${executionId}/${request.payload.name}`
+                const instance = WorkflowInstance.initial(workflow, executionId)
+                return Effect.gen(function*() {
                   let entry = activities.get(activityId)
                   while (!entry) {
                     const latch = Effect.unsafeMakeLatch()
                     activityLatches.set(activityId, latch)
                     yield* latch.await
                     entry = activities.get(activityId)
                   }
-                  const instance = WorkflowInstance.initial(workflow, executionId)
                   const contextMap = new Map(entry.runtime.context.unsafeMap)
                   contextMap.set(Activity.CurrentAttempt.key, request.payload.attempt)
                   contextMap.set(WorkflowInstance.key, instance)
@@ -311,23 +313,29 @@ export const make = Effect.gen(function*() {
                     runtimeFlags: Runtime.defaultRuntimeFlags
                   })
                   return yield* entry.activity.executeEncoded.pipe(
-                    Effect.interruptible,
-                    Effect.onInterrupt(() => {
-                      instance.suspended = true
-                      return Effect.void
-                    }),
-                    Workflow.intoResult,
-                    Effect.provide(runtime),
-                    Effect.ensuring(Effect.sync(() => {
-                      activities.delete(activityId)
-                    }))
+                    Effect.provide(runtime)
                   )
-                },
-                Rpc.wrap({
-                  fork: true,
-                  uninterruptible: true
-                })
-              ),
+                }).pipe(
+                  Workflow.intoResult,
+                  Effect.catchAllCause((cause) => {
+                    const interruptors = Cause.interruptors(cause)
+                    // we only want to store explicit interrupts
+                    const ids = Array.from(interruptors, (id) => Array.from(FiberId.ids(id))).flat()
+                    const suspend = ids.includes(RpcServer.fiberIdClientInterrupt.id) ||
+                      ids.includes(RpcServer.fiberIdTransientInterrupt.id)
+                    return suspend ? Effect.succeed(new Workflow.Suspended()) : Effect.failCause(cause)
+                  }),
+                  Effect.provideService(WorkflowInstance, instance),
+                  Effect.provideService(Activity.CurrentAttempt, request.payload.attempt),
+                  Effect.ensuring(Effect.sync(() => {
+                    activities.delete(activityId)
+                  })),
+                  Rpc.wrap({
+                    fork: true,
+                    uninterruptible: true
+                  })
+                )
+              },
 
               deferred: Effect.fnUntraced(function*(request: Entity.Request<any>) {
                 yield* ensureSuccess(resume(workflow, executionId))

packages/cluster/src/EntityAddress.ts

@@ -63,3 +63,13 @@ export class EntityAddress extends Schema.Class<EntityAddress>(SymbolKey)({
 export const EntityAddressFromSelf: Schema.Schema<EntityAddress> = Schema.typeSchema(
   EntityAddress
 )
+
+/**
+ * @since 4.0.0
+ * @category constructors
+ */
+export const make = (options: {
+  readonly shardId: ShardId
+  readonly entityType: EntityType
+  readonly entityId: EntityId
+}): EntityAddress => new EntityAddress(options, { disableValidation: true })

packages/cluster/src/EntityId.ts

@@ -14,3 +14,9 @@ export const EntityId = Schema.NonEmptyTrimmedString.pipe(Schema.brand("EntityId
  * @category models
  */
 export type EntityId = typeof EntityId.Type
+
+/**
+ * @since 1.0.0
+ * @category constructors
+ */
+export const make = (id: string): EntityId => id as EntityId