From c6eeb7b8bcb7169bc30d85b2bbbeb53cc57b9517 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Julien=20Valverd=C3=A9?= Date: Thu, 23 Jul 2026 03:08:42 +0200 Subject: [PATCH] Fix --- packages/effect-fc-next/src/ScopeRegistry.ts | 55 +++++++++----------- 1 file changed, 26 insertions(+), 29 deletions(-) diff --git a/packages/effect-fc-next/src/ScopeRegistry.ts b/packages/effect-fc-next/src/ScopeRegistry.ts index 2cb0ffd..643f1d8 100644 --- a/packages/effect-fc-next/src/ScopeRegistry.ts +++ b/packages/effect-fc-next/src/ScopeRegistry.ts @@ -1,4 +1,4 @@ -import { type Cause, Context, DateTime, type Duration, Effect, Equal, Exit, HashMap, Layer, Option, Order, Scope, Semaphore, Stream, SubscriptionRef } from "effect" +import { type Cause, Context, DateTime, type Duration, Effect, Equal, Exit, flow, HashMap, Layer, Option, Order, Scope, Semaphore, Stream, SubscriptionRef } from "effect" export interface ScopeRegistryService { @@ -39,14 +39,13 @@ export class ScopeRegistryServiceImpl implements ScopeRegistryService { readonly runSemaphore: Semaphore.Semaphore, ) {} - register( key: ScopeRegistryService.Key, options: ScopeRegistryService.RegisterOptions, ): Effect.Effect { return Effect.gen({ self: this }, function*() { const now = yield* DateTime.now - const entry = Equal.byReference({ + const entry = Equal.byReference({ scope: yield* Scope.make(options.finalizerExecutionStrategy), expiresAt: Option.some(DateTime.addDuration(now, options.finalizerExecutionDebounce)), finalizerExecutionDebounce: options.finalizerExecutionDebounce, @@ -58,21 +57,34 @@ export class ScopeRegistryServiceImpl implements ScopeRegistryService { } commit(key: ScopeRegistryService.Key): Effect.Effect { - return this.updateEntry(key, entry => Equal.byReference({ - ...entry, - expiresAt: Option.none(), - })) + return SubscriptionRef.get(this.ref).pipe( + Effect.map(HashMap.get(key)), + Effect.flatMap(Effect.fromOption), + Effect.map(entry => Equal.byReference({ + ...entry, + expiresAt: Option.none(), + })), + Effect.tap(entry => SubscriptionRef.update(this.ref, HashMap.set(key, entry))), + ) } release(key: ScopeRegistryService.Key): Effect.Effect { - return Effect.flatMap(DateTime.now, now => this.updateEntry(key, entry => Equal.byReference({ - ...entry, - expiresAt: Option.some(DateTime.addDuration(now, entry.finalizerExecutionDebounce)), - }))) + return SubscriptionRef.get(this.ref).pipe( + Effect.map(HashMap.get(key)), + Effect.flatMap(option => Effect.all([ + DateTime.now, + Effect.fromOption(option), + ])), + Effect.map(([now, entry]) => Equal.byReference({ + ...entry, + expiresAt: Option.some(DateTime.addDuration(now, entry.finalizerExecutionDebounce)), + })), + Effect.tap(entry => SubscriptionRef.update(this.ref, HashMap.set(key, entry))), + ) } get run(): Effect.Effect { - return this.runSemaphore.withPermits(1)(SubscriptionRef.changes(this.ref).pipe( + return SubscriptionRef.changes(this.ref).pipe( Stream.switchMap(entries => Option.match(this.getNextExpiration(entries), { onNone: () => Stream.never, onSome: expiresAt => Stream.fromEffect(DateTime.now.pipe( @@ -83,7 +95,8 @@ export class ScopeRegistryServiceImpl implements ScopeRegistryService { )), })), Stream.runDrain, - )) + this.runSemaphore.withPermit, + ) } get dispose(): Effect.Effect { @@ -99,22 +112,6 @@ export class ScopeRegistryServiceImpl implements ScopeRegistryService { ) } - - updateEntry( - key: ScopeRegistryService.Key, - f: (entry: ScopeRegistryService.Entry) => ScopeRegistryService.Entry, - ): Effect.Effect { - return SubscriptionRef.modify(this.ref, entries => { - const current = HashMap.get(entries, key) - - if (current._tag === "None") - return [Option.none(), entries] - - const entry = f(current.value) - return [Option.some(entry), HashMap.set(entries, key, entry)] - }).pipe(Effect.flatMap(Effect.fromOption)) - } - get closeExpired(): Effect.Effect { return Effect.flatMap(DateTime.now, now => SubscriptionRef.modify( this.ref,