Locks | tRPC

experimental_lock() runs the rest of the call under a lock, so calls with the same key never overlap: the second waits for the first, and fails with experimental_LockConflict (CONFLICT, 409) if timeout (10 seconds) passes first. The resolver gets ctx.lock.waited.

import { import NodeRedisNodeRedis } from '@effect/platform-node';
import {
  function experimental_lock<TReq extends MiddlewareRequirements = {}>(options: experimental_LockMiddlewareOptions<ReqCtx<TReq>, ReqInput<TReq>, ReqMeta<TReq>>): EffectMiddleware<ReqCtx<TReq>, ReqInput<TReq>, ReqMeta<TReq>, {
    readonly lock: experimental_Lock;
}, experimental_LockConflict, experimental_Locker, never, never, Unset>experimental_lock,
  function experimental_redisLocker(options?: experimental_RedisLockerOptions): Layer.Layer<experimental_Locker, never, Redis>experimental_redisLocker,
} from '@trpc12/trpc/middleware';

const const pay: Procedure<{
    type: "mutation";
    ctx: {};
    meta: {};
    input: {
        invoice: string;
    };
    output: void;
    errors: experimental_LockConflict | SchemaValidationError<z.ZodObject<{
        invoice: z.ZodString;
    }, z.core.$strip>>;
    services: experimental_Locker;
}>pay = const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure
  .ProcedureBuilder<RootDef<{}, {}>>.input<z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>(schema: z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>): ProcedureBuilder<InputDef<RootDef<{}, {}>, z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>> (+1 overload)input(import zz.function object<{
    invoice: z.ZodString;
}>(shape?: {
    invoice: z.ZodString;
} | undefined, params?: string | {
    error?: string | z.core.$ZodErrorMap<NonNullable<z.core.$ZodIssueInvalidType<unknown> | z.core.$ZodIssueUnrecognizedKeys>> | undefined;
    message?: string | undefined | undefined;
} | undefined): z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>object({ invoice: z.ZodStringinvoice: import zz.function string(params?: string | z.core.$ZodStringParams): z.ZodString (+1 overload)string() }))
  .ProcedureBuilder<InputDef<RootDef<{}, {}>, ZodObject<{ invoice: ZodString; }, $strip>>>.use<{
    readonly lock: experimental_Lock;
}, experimental_LockConflict, experimental_Locker, never, never, Unset, {}>(mw: EffectMiddleware<{}, {
    invoice: string;
}, {}, {
    readonly lock: experimental_Lock;
}, experimental_LockConflict, experimental_Locker, never, never, Unset>): ProcedureBuilder<With<InputDef<RootDef<{}, {}>, z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>, {
    ...;
}>> (+1 overload)use(
    experimental_lock<{
    input: {
        invoice: string;
    };
}>(options: experimental_LockMiddlewareOptions<{}, {
    invoice: string;
}, {}>): EffectMiddleware<{}, {
    invoice: string;
}, {}, {
    readonly lock: experimental_Lock;
}, experimental_LockConflict, experimental_Locker, never, never, Unset>experimental_lock<{ input: {
    invoice: string;
}input: { invoice: stringinvoice: string } }>({
      experimental_LockMiddlewareOptions<{}, { invoice: string; }, {}>.key: (opts: {
    readonly ctx: {};
    readonly input: {
        invoice: string;
    };
    readonly meta: {};
    readonly path: string;
}) => stringkey: (opts: {
    readonly ctx: {};
    readonly input: {
        invoice: string;
    };
    readonly meta: {};
    readonly path: string;
}opts) => `invoice:${opts: {
    readonly ctx: {};
    readonly input: {
        invoice: string;
    };
    readonly meta: {};
    readonly path: string;
}opts.input: {
    invoice: string;
}input.invoice: stringinvoice}`,
    }),
  )
  .ProcedureBuilder<With<InputDef<RootDef<{}, {}>, ZodObject<{ invoice: ZodString; }, $strip>>, { ctx: {}; ctxAdded: { readonly lock: experimental_Lock; }; ... 4 more ...; requires: experimental_Locker; }>>.mutation<void>(resolver: (opts: ResolverOpts<With<InputDef<RootDef<{}, {}>, z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>, {
    ctx: {};
    ctxAdded: {
        readonly lock: experimental_Lock;
    };
    inputOut: {
        invoice: string;
    };
    errors: experimental_LockConflict | SchemaValidationError<z.ZodObject<{
        invoice: z.ZodString;
    }, z.core.$strip>>;
    shorts: never;
    provided: Response;
    requires: experimental_Locker;
}>>) => void): Procedure<...>mutation((opts: ResolverOpts<With<InputDef<RootDef<{}, {}>, z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>, {
    ctx: {};
    ctxAdded: {
        readonly lock: experimental_Lock;
    };
    inputOut: {
        invoice: string;
    };
    errors: experimental_LockConflict | SchemaValidationError<z.ZodObject<{
        invoice: z.ZodString;
    }, z.core.$strip>>;
    shorts: never;
    provided: Response;
    requires: experimental_Locker;
}>>opts) =>
    function charge(invoice: string, opts: {
    retried: boolean;
}): voidcharge(opts: ResolverOpts<With<InputDef<RootDef<{}, {}>, z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>, {
    ctx: {};
    ctxAdded: {
        readonly lock: experimental_Lock;
    };
    inputOut: {
        invoice: string;
    };
    errors: experimental_LockConflict | SchemaValidationError<z.ZodObject<{
        invoice: z.ZodString;
    }, z.core.$strip>>;
    shorts: never;
    provided: Response;
    requires: experimental_Locker;
}>>opts.ResolverOpts<With<InputDef<RootDef<{}, {}>, ZodObject<{ invoice: ZodString; }, $strip>>, { ctx: {}; ctxAdded: { readonly lock: experimental_Lock; }; ... 4 more ...; requires: experimental_Locker; }>>.input: {
    invoice: string;
}input.invoice: stringinvoice, { retried: booleanretried: opts: ResolverOpts<With<InputDef<RootDef<{}, {}>, z.ZodObject<{
    invoice: z.ZodString;
}, z.core.$strip>>, {
    ctx: {};
    ctxAdded: {
        readonly lock: experimental_Lock;
    };
    inputOut: {
        invoice: string;
    };
    errors: experimental_LockConflict | SchemaValidationError<z.ZodObject<{
        invoice: z.ZodString;
    }, z.core.$strip>>;
    shorts: never;
    provided: Response;
    requires: experimental_Locker;
}>>opts.ResolverOpts<With<InputDef<RootDef<{}, {}>, ZodObject<{ invoice: ZodString; }, $strip>>, { ctx: {}; ctxAdded: { readonly lock: experimental_Lock; }; ... 4 more ...; requires: experimental_Locker; }>>.ctx: {
    readonly lock: experimental_Lock;
}ctx.lock: experimental_Locklock.experimental_Lock.waited: booleanwaited }),
  );

createFetchHandler<{
    pay: typeof pay;
}, undefined>(opts: FetchHandlerOptions<{
    pay: typeof pay;
}, undefined>): FetchHandler<undefined>createFetchHandler({
  BaseHandlerOptions<{ pay: typeof pay; }>.router: {
    pay: typeof pay;
}router,
  layer: Layer.Layer<experimental_Locker, any, never>layer: function experimental_redisLocker(options?: experimental_RedisLockerOptions): Layer.Layer<experimental_Locker, never, Redis>experimental_redisLocker().Pipeable.pipe<Layer.Layer<experimental_Locker, never, Redis>, Layer.Layer<experimental_Locker, RedisError, never>>(this: Layer.Layer<experimental_Locker, never, Redis>, ab: (_: Layer.Layer<experimental_Locker, never, Redis>) => Layer.Layer<experimental_Locker, RedisError, never>): Layer.Layer<experimental_Locker, RedisError, never> (+21 overloads)pipe(
    import LayerLayer.const provide: <never, RedisError, Redis | NodeRedis.NodeRedis>(that: Layer.Layer<Redis | NodeRedis.NodeRedis, RedisError, never>) => <RIn2, E2, ROut2>(self: Layer.Layer<ROut2, E2, RIn2>) => Layer.Layer<ROut2, RedisError | E2, Exclude<RIn2, Redis | NodeRedis.NodeRedis>> (+3 overloads)provide(import NodeRedisNodeRedis.const layer: (options?: RedisClientOptions<RedisModules, RedisFunctions, RedisScripts, RespVersions, TypeMapping, RedisSocketOptions> | undefined) => Layer.Layer<Redis | NodeRedis.NodeRedis, RedisError>layer({ RedisClientOptions<M extends RedisModules = RedisModules, F extends RedisFunctions = RedisFunctions, S extends RedisScripts = RedisScripts, RESP extends RespVersions = 3, TYPE_MAPPING extends TypeMapping = TypeMapping, SocketOptions extends RedisSocketOptions = RedisSocketOptions>.url?: string | undefinedurl: var process: NodeJS.Processprocess.NodeJS.Process.env: NodeJS.ProcessEnvenv.string | undefinedREDIS_URL })),
  ),
});

A holder loses the lock after ttl (30 seconds), so a crashed instance can't hold it forever. A key the call already holds, from an earlier experimental_lock() in its chain, isn't taken twice. The lock covers the resolver; a streamed output or a subscription keeps running after it's released.

Lockers

The locker is an Effect service, experimental_Locker, provided as a layer:

experimental_memoryLocker() for one process.
experimental_redisLocker() over Effect's Redis for many. Any Redis Effect's Redis.make({ send, subscribe }) can wrap works, Upstash's REST client included.
experimental_durableLocker() on Workers.

Outside a procedure, take a lock with the service: yield* (yield* experimental_Locker).acquire(key) holds it until the surrounding Effect.scoped ends.

The examples/experimental/durable-jobs app takes an experimental_redisLocker() lock per invoice, so two payments for one invoice on two servers run one at a time.

Credits

experimental_lock() follows oRPC's lock helpers, and its Durable Object store, experimental_durableLocker(), works like oRPC's.