add Tracing

package.json

@@ -1,7 +1,7 @@
 {
   "private": true,
   "type": "module",
-  "packageManager": "pnpm@9.5.0",
+  "packageManager": "pnpm@10.10.0+sha512.d615db246fe70f25dcfea6d8d73dee782ce23e2245e3c4f6f888249fb568149318637dca73c2c5c8ef2a4ca0d5657fb9567188bfab47f566d1ee6ce987815c39",
   "workspaces": [
     "packages/*"
   ],
@@ -48,7 +48,7 @@
     "babel-plugin-annotate-pure-calls": "^0.5.0",
     "effect": "^3.14.16",
     "eslint": "^8.57.1",
-    "eslint-import-resolver-typescript": "^3.10.0",
+    "eslint-import-resolver-typescript": "^3.10.1",
     "eslint-plugin-codegen": "^0.30.0",
     "eslint-plugin-deprecation": "^3.0.0",
     "eslint-plugin-import": "^2.31.0",

packages/core/package.json

@@ -40,13 +40,15 @@
   "dependencies": {
     "@parcel/watcher": "^2.5.1",
     "esbuild": "^0.25.3",
-    "typescript": "^5.8.3"
+    "typescript": "^5.8.3",
+    "undici": "^7.8.0"
   },
   "peerDependencies": {
     "effect": "^3.14"
   },
   "devDependencies": {
     "@effect/cli": "^0.59.16",
+    "@effect/opentelemetry": "^0.46.13",
     "@effect/platform": "^0.80.16",
     "@effect/platform-node": "^0.77.4",
     "@effect/rpc": "^0.56.4",

packages/core/src/ContentWorker.ts

@@ -1,7 +1,9 @@
 /**
  * @since 1.0.0
  */
+import * as OtlpTracer from "@effect/opentelemetry/OtlpTracer"
 import * as NodeContext from "@effect/platform-node/NodeContext"
+import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient"
 import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
 import * as NodeWorkerRunner from "@effect/platform-node/NodeWorkerRunner"
 import type { WorkerError } from "@effect/platform/WorkerError"
@@ -24,6 +26,8 @@ import * as Source from "./Source.ts"
 
 const Handlers = ContentWorkerSchema.Rpcs.toLayer(Effect.gen(function*() {
   const storage = yield* DocumentStorage
+  const workerSpan = yield* Effect.makeSpanScoped("ContentWorker.Handlers")
+  const semaphore = yield* Effect.makeSemaphore(2)
 
   const configs = yield* RcMap.make({
     lookup: Effect.fnUntraced(function*(path: ContentWorkerSchema.ConfigPath) {
@@ -43,6 +47,10 @@ const Handlers = ContentWorkerSchema.Rpcs.toLayer(Effect.gen(function*() {
             Stream.orDie,
             Stream.mapEffect(
               Effect.fnUntraced(function*(event) {
+                yield* Effect.annotateCurrentSpan({
+                  event: event._tag,
+                  eventId: event.id
+                })
                 if (event._tag !== "Added") return
                 const { output } = event
                 const decoded = yield* (Schema.decode(document.fields)(output.fields) as Effect.Effect<
@@ -81,6 +89,7 @@ const Handlers = ContentWorkerSchema.Rpcs.toLayer(Effect.gen(function*() {
             ),
             Effect.forever,
             Effect.provideService(Source.WorkerEventStream, Mailbox.toStream(mailbox)),
+            Effect.withParentSpan(workerSpan),
             Effect.forkScoped,
             Effect.interruptible
           )
@@ -103,7 +112,15 @@ const Handlers = ContentWorkerSchema.Rpcs.toLayer(Effect.gen(function*() {
       const process = (name: string, id: string, meta: unknown) =>
         RcMap.get(docProcessor, name).pipe(
           Effect.flatMap((process) => process(id, meta)),
-          Effect.scoped
+          Effect.scoped,
+          Effect.withSpan("ContentWorker.process", {
+            parent: workerSpan,
+            attributes: {
+              id,
+              meta
+            }
+          }),
+          semaphore.withPermits(1)
         )
 
       return { ...config, process } as const
@@ -148,14 +165,22 @@ const resolveComputedFields = (options: {
       )
   ) as Effect.Effect<Record<string, unknown>, ParseError>
 
+const TracerLayer = OtlpTracer.layer({
+  url: "http://localhost:4318/v1/traces",
+  resource: {
+    serviceName: "@effect/contentlayer/ContentWorker"
+  }
+}).pipe(Layer.provide(NodeHttpClient.layerUndici))
+
 /**
  * @since 1.0.0
  * @category Layers
  */
 export const layer: Layer.Layer<never, WorkerError> = RpcServer.layer(ContentWorkerSchema.Rpcs).pipe(
   Layer.provide(Handlers),
   Layer.provide(RpcServer.layerProtocolWorkerRunner),
-  Layer.provide(NodeWorkerRunner.layer)
+  Layer.provide(NodeWorkerRunner.layer),
+  Layer.provide(TracerLayer)
 )
 
 Layer.launch(layer).pipe(

packages/core/src/DocumentBuilder.ts

@@ -130,8 +130,15 @@ export const run = Effect.gen(function*() {
             ),
             Stream.let("output", ({ event }) => event.output),
             Stream.mapEffect(
-              Effect.fnUntraced(
+              Effect.fn("DocumentBuilder.process")(
                 function*({ document, event, output }) {
+                  yield* Effect.annotateCurrentSpan({
+                    documentType: document.name,
+                    documentId: output.id,
+                    event: event._tag,
+                    eventInitial: event.initial
+                  })
+
                   const cached = event.initial && cache.exists(output.id, output.version)
                   if (!cached) {
                     yield* worker.ProcessDocument({
@@ -190,6 +197,7 @@ export const run = Effect.gen(function*() {
       { switch: true }
     ),
     Stream.runDrain,
+    Effect.withSpan("DocumentBuilder.run"),
     BuildError.catchAndLog,
     Effect.catchAllCause(Effect.logError)
   )
@@ -207,7 +215,7 @@ export class ContentWorkerPool
         minSize: Math.max(1, Math.floor(workerPoolSize / 2)),
         maxSize: workerPoolSize,
         timeToLive: "30 seconds",
-        concurrency: 10
+        concurrency: 15
       }).pipe(
         Layer.provide(NodeWorker.layerPlatform(() => tsWorker("./ContentWorker")))
       )

packages/core/src/SourcePlugin.ts

@@ -36,35 +36,28 @@ export const make = <E2, Meta, Context, Context2>(
       return self.events as Stream.Stream<Source.Event<Meta, Context2>, E | E2>
     }
 
-    const latch = yield* Effect.makeLatch(true)
     const mailbox = yield* Mailbox.make<Source.Output<Meta, Context>>()
     const events = yield* Mailbox.make<Source.Event<Meta, Context2>, E | E2>()
     let initial = true
 
     yield* self.events.pipe(
-      Stream.mapEffect(Effect.fnUntraced(function*(event) {
-        yield* latch.await
+      Stream.mapEffect((event) => {
         if (event._tag === "Removed") {
-          return yield* events.offer(event)
+          return events.offer(event)
         }
         if (initial && !event.initial) {
           initial = false
         }
-        yield* mailbox.offer(event.output)
-        latch.unsafeClose()
-      })),
+        return mailbox.offer(event.output)
+      }),
       Stream.runDrain,
       Effect.onExit((exit) => exit._tag === "Success" ? mailbox.end : events.failCause(exit.cause)),
       Effect.interruptible,
       Effect.forkScoped
     )
 
     yield* f(Mailbox.toStream(mailbox)).pipe(
-      Stream.runForEach((output) =>
-        events.offer(Source.EventAdded(output, initial)).pipe(
-          Effect.andThen(latch.open)
-        )
-      ),
+      Stream.runForEach((output) => events.offer(Source.EventAdded(output, initial))),
       Mailbox.into(events),
       Effect.interruptible,
       Effect.forkScoped

packages/core/src/bin.ts

@@ -1,14 +1,14 @@
 #!/usr/bin/env node
-import * as DevTools from "@effect/experimental/DevTools"
 import * as NodeContext from "@effect/platform-node/NodeContext"
 import * as ParcelWatcher from "@effect/platform-node/NodeFileSystem/ParcelWatcher"
 import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
 import * as Effect from "effect/Effect"
 import * as Layer from "effect/Layer"
 import * as Cli from "./Cli.ts"
+import { TracerLayer } from "./internal/Tracing.ts"
 
 const MainLive = Layer.mergeAll(
-  DevTools.layer(),
+  TracerLayer,
   NodeContext.layer.pipe(
     Layer.provide(ParcelWatcher.layer)
   )

packages/core/src/internal/Tracing.ts

@@ -0,0 +1,10 @@
+import * as OtlpTracer from "@effect/opentelemetry/OtlpTracer"
+import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient"
+import * as Layer from "effect/Layer"
+
+export const TracerLayer = OtlpTracer.layer({
+  url: "http://localhost:4318/v1/traces",
+  resource: {
+    serviceName: "@effect/contentlayer"
+  }
+}).pipe(Layer.provide(NodeHttpClient.layerUndici))