Data source agnostic data loader with support for tiered in-memory and async caching, fetch deduplication and fallback data sources. Implements Cache-Aside, Read-Through and Refresh-Ahead patterns.

Special thanks to Diana Baužytė for creating the project logo.

You can watch NodeConf EU 2023 talk for a brief and visual overview of what new features layered-loader brings to the table of the Node.js caching.

Contents

• Prerequisites
• Entrypoints
• Use-cases
• Feature Comparison
• Performance Comparison
  • In-Memory Store
  • Redis Store
• Basic concepts
• Basic example
  • Simplified loader syntax
• Loader API
• Parametrized loading
• Update notifications
  • Available notification adapters
  • Picking a notification adapter
  • Redis pub/sub
  • AWS SNS/SQS
    • How fanout works
• Flexible invalidation triggers
  • Recommended pattern: Redis fanout + SQS trigger
  • SNS/SQS adapter
• Applying invalidations from your own transport
• Background work
• Isolate runtimes with no invalidation bus
  • Hoist the loader to isolate scope
  • Pull invalidations instead of pushing them
  • Hand background work to the request
  • Validating on every read
  • What the library does not do
• Cache statistics
• Cache-only operations
  • Forcing an update
  • Forcing a specific value
• Usage in high-performance systems
  • Synchronous short-circuit
  • Preemptive background refresh
  • Conditional refresh with a staleness check
• Group operations
• Provided async caches
  • RedisCache
• Redis connection safety
  • Automatic READONLY reconnection
  • Using enrichRedisConfig for your own connections
  • Cloud-optimized configuration

Prerequisites

Node: 22+

layered-loader is an ESM-only package. It is published as ECMAScript modules with an exports map and has no CommonJS build, so it is consumed with import from ESM code (or with a dynamic await import('layered-loader') from CommonJS). Only the entrypoints listed below and package.json are exported — reaching into layered-loader/dist/... is not supported.

Entrypoints

The package is split into two subpath entrypoints so that consumers who do not run Redis never pull ioredis into their module graph, or into the types they compile against:

| Entrypoint | Contents | Can reach ioredis? | | ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | | layered-loader/core | Loader, GroupLoader, ManualCache, ManualGroupCache, the in-memory caches, AbstractNotificationConsumer, key resolvers, all types | Never | | layered-loader/redis | RedisCache, RedisGroupCache, createNotificationPair, createGroupNotificationPair, the Redis publishers/consumers, enrichRedisConfig | Only when you construct a client from connection options | | layered-loader | Everything from both of the above | Same as /redis |

The root entrypoint is a plain export * of both, so it remains a superset of the other two and existing imports keep working unchanged.

If your infrastructure has no Redis — or you are targeting a runtime that cannot load a Node-only client at all, such as Cloudflare Workers — import from layered-loader/core:

import { Loader, type InMemoryCacheConfiguration } from 'layered-loader/core'

const loader = new Loader<string>({
  inMemoryCache: { ttlInMsecs: 60_000 } satisfies InMemoryCacheConfiguration,
  dataSourceGetOneFn: (key) => fetchFromApi(key),
})

Nothing reachable from layered-loader/core imports ioredis, or even a node: builtin — its only runtime dependency is toad-cache. Importing cleanly is necessary but not sufficient on an isolate runtime, though: see Isolate runtimes with no invalidation bus for how to keep the cache alive across invocations, invalidate it without a bus, and stop background work from outliving its request.

Redis usage keeps its own entrypoint:

import { RedisCache } from 'layered-loader/redis'

How optional ioredis really is

ioredis is an optional peer dependency: it is not installed unless you ask for it, and nothing in the package touches it — as a value or as a type — until you construct a client from connection options.

Add it to your own package.json only if you use Redis:

// your package.json — only needed if you use the Redis entrypoint
"dependencies": {
  "ioredis": "^6.0.0"
}

What that buys you, precisely

Four different things get conflated under "optional". All four hold here, and they are worth separating:

| | layered-loader/core | layered-loader/redis and the root | | ------------------------------------------ | --------------------- | ----------------------------------- | | Installed unless you ask for it | No | No | | Loaded when you import the entrypoint | No | No — resolved lazily | | Present in the types you compile against | No | No — vendored structurally | | Needed at runtime | Never | Only to construct a client from connection options |

So every entrypoint imports cleanly with ioredis absent from node_modules entirely, and layered-loader type-checks under skipLibCheck: false without it. The surface this package uses is vendored as structural interfaces in lib/redis/RedisLike.ts, and a CI type-check asserts they still match the real ioredis declarations, so they cannot drift silently.

When you do need it installed

You need ioredis in your own dependencies if either of these is true:

• You use RedisCache or RedisGroupCache. These take a client you construct yourself, so you are importing ioredis directly anyway.

• You call createNotificationPair / createGroupNotificationPair with plain connection options rather than with a client. That constructs a client internally, and throws an actionable error if ioredis is missing:

  Constructing a Redis client from connection options requires the optional peer dependency "ioredis" to be installed. Install it, or pass an already-constructed client instead.

  Passing an already-constructed client avoids that path entirely.

Bundling

Because ioredis is reached through a lazy require, bundlers cannot see it and will not inline it. That is the right behaviour for an optional peer — it is your dependency, not this package's — but it has a consequence worth knowing before you deploy: if you bundle an app that constructs clients from connection options, ioredis must still be resolvable at runtime. Keep it in node_modules next to the bundle, or add it to your bundler's input explicitly. Bundling layered-loader/core needs none of this.

Verifying it yourself

To confirm that no part of your own app drags ioredis in, run this against your entrypoint — it records every specifier Node actually resolves, so it sees static imports, dynamic import() and require() alike:

import { registerHooks } from 'node:module'

const loaded = new Set()
registerHooks({
  resolve(specifier, context, nextResolve) {
    loaded.add(specifier)
    return nextResolve(specifier, context)
  },
})

await import('./your-app.js')

const redis = [...loaded].filter((specifier) => specifier.includes('ioredis'))
console.log(redis.length ? `loaded: ${redis.join(', ')}` : 'ioredis was never loaded')

Cache invalidation without Redis

If you reached for Redis only to invalidate caches across instances, you do not need it at all: @layered-loader/sqs provides the same notification publisher/consumer pair over SNS/SQS, and imports from layered-loader/core.

Use-cases

This library has four main goals:

1. Provide transparent, high performance, flexible caching mechanism for data retrieval operations;
2. Prevent redundant data retrieval in high-load systems;
3. Support distributed in-memory cache invalidation to prevent stale data in cache;
4. Enable fallback mechanism for retrieving data when alternate sources exist;

Feature Comparison

Since there are a few cache solutions, here is a table comparing them:

| Feature | layered-loader | async-cache-dedupe | dataloader | cache-manager | | :----------------------------------------------- | :-----------------------------------------------------------: | :------------------------------------------------------------------: | :-------------------------------------------------: | :-----------------------------------------------------------------------: | | Single Entity Fetch | ✓ | ✓ | ✓ | ✓ | | Bulk Entity Fetch | ✓ | | ✓ | ✓ | | Single Entity Fetch Deduplication (Read-Through) | ✓ | ✓ | ✓ | | | Bulk Entity Fetch Deduplication | ✓ | | ✓ | | | Preemptive Cache Refresh (Refresh-Ahead) | ✓ | | | | | Tiered Caches | ✓ | | | ✓ | | Group Support | ✓ | partially, references for invalidation | | | | Redis Support | ✓ | ✓ | | ✓ | | Redis Key Auto-Prefixing | ✓ | | | | | Synchronous In-Memory Cache Access | ✓ | | | | | Distributed In-Memory Cache Invalidation | ✓ | | | | | Hit/Miss/Expiration Tracking | ✓ | partially, hooks available | | | | Support For Custom Cache Stores | ✓ | | | ✓ | | Optimized for | Broad‑Scope Use | Single Entity Fetch Deduplication | Bulk Entity Fetch Deduplication | Manual Caching |

Performance Comparison

You can find all the benchmarks used for the comparison in NodeJS benchmark repo. Please let us know if they can be made more accurate!

In-Memory Store

Higher is better:

| Feature - Ops/sec | layered-loader | async-cache-dedupe | dataloader | cache-manager | toad-cache | tiny-lru | | :----------------------------- | :-----------------------------------------------------------: | :------------------------------------------------------------------: | :-------------------------------------------------: | :-----------------------------------------------------------------------: | :---------------------------------------------------: | :-----------------------------------------------: | | Single Entity Fetch | 3836.436 | 446.146 | 717.420 | ToDo | 4191.279 | 3818.146 | | Bulk Entity Fetch | | | | | | | | Concurrent Single Entity Fetch | | | | | | | | Concurrent Bulk Entity Fetch | | | | | | |

Redis Store

Higher is better:

| Feature - Ops/sec | layered-loader | async-cache-dedupe | cache-manager | ioredis | | :----------------------------- | :-----------------------------------------------------------: | :------------------------------------------------------------------: | :-----------------------------------------------------------------------: | :-----------------------------------------: | | Single Entity Fetch | | | | | | Bulk Entity Fetch | | | | | | Concurrent Single Entity Fetch | 167.745 | 124.854 | 40.234 | 47.775 | | Concurrent Bulk Entity Fetch | | | | |

Basic concepts

There are two main entity types defined by layered-loader:

1. Loader - defined procedure of retrieving data from one or more data sources with full deduplication (same resource is only asked once at any given time), with an optional caches in the middle. Loader is composed of Data Sources and Caches.
2. Manual cache - async cache and/or sync in-memory cache, with deduplication for retrieval commands, which is populated explicitly.

Loaders and caches are composed out of the following building blocks.

1. InMemoryCache - synchronous in-memory cache. Offers highest possible performance. If used with a longer TTL, you should consider using a notification Publisher/Consumer pair for distributed cache invalidation, to prevent your cached data from becoming stale;
2. AsyncCache - asynchronous remote cache. Slower than in-memory cache, but can be invalidated more easily, as it is shared across all nodes of a distributed system.
3. Data Source - primary source of truth of data, that can be used for populating caches. Used in a strictly read-only mode.

• layered-loader will try loading the data from the data source defined for the Loader, in the following order: InMemory, AsyncCache, DataSources. In case undefined value is the result of retrieval, next source in sequence will be used, until there is either a value, or there are no more sources available;
• null and undefined have different semantics:
  • null means "value was successfully resolved, but it is empty" - this will be cached and subsequent data sources will not be queried;
  • undefined means "value was not resolved" - this will NOT be cached and the next data source in the sequence will be queried. If all data sources return undefined, the Loader returns undefined without caching anything;
• If non-last data source throws an error, it is handled using configured ErrorHandler. If the last data source throws an error, and there are no remaining fallback data sources, an error will be thrown by the Loader.
• If any caches (InMemoryCache or AsyncCache) precede the source, that returned a value, all of them will be updated with that value;
• If there is an ongoing retrieval operation for the given key, promise for that retrieval will be reused and returned as a result of loader.get, instead of starting a new retrieval.
• You can use just the memory cache, just the asynchronous one, neither, or both. Unconfigured layer will be simply skipped for all operations (both storage and retrieval).

Basic example

Let's define a data source, which will be the primary source of truth, and two levels of caching:

import Redis from 'ioredis'
import { RedisCache, InMemoryCache } from 'layered-loader'
import type { DataSource } from 'layered-loader'

const ioRedis = new Redis({
  host: 'localhost',
  port: 6379,
  password: 'sOmE_sEcUrE_pAsS',
})

class ClassifiersDataSource implements DataSource<Record<string, any>> {
  private readonly db: Knex
  name = 'Classifiers DB loader'
  isCache = false

  constructor(db: Knex) {
    this.db = db
  }

  async get(key: string): Promise<Record<string, any> | undefined | null> {
    const results = await this.db('classifiers')
      .select('*')
      .where({
        id: parseInt(key),
      })
    return results[0]
  }

  async getMany(keys: string[]): Promise<Record<string, any>[]> {
    return this.db('classifiers').select('*').whereIn('id', keys.map(parseInt))
  }
}

const loader = new Loader<string>({
  // this cache will be checked first
  inMemoryCache: {
    cacheType: 'lru-map', // you can choose between lru and fifo caches, fifo being 10% slightly faster
                          // 'lru-object' is another option, it is slightly faster for non-string keys
    ttlInMsecs: 1000 * 60,
    maxItems: 100,
  },

  // this cache will be checked if in-memory one returns undefined
  asyncCache: new RedisCache<string>(ioRedis, {
    json: true, // this instructs loader to serialize passed objects as string and deserialize them back to objects
    ttlInMsecs: 1000 * 60 * 10,
  }),

  // this will be used if neither cache has the requested data
  dataSources: [new ClassifiersDataSource(db)],
})

// If cache is empty, but there is data in the DB, after this operation is completed, both caches will be populated
const classifier = await loader.get('1')

Simplified loader syntax

It is also possible to inline datasource definition:

const loader = new Loader<string>({
  // this cache will be checked first
  inMemoryCache: {
    cacheType: 'lru-map', // you can choose between lru and fifo caches, fifo being 10% slightly faster
                          // 'lru-object' is another option, it is slightly faster for non-string keys
    ttlInMsecs: 1000 * 60,
    maxItems: 100,
  },

  // this cache will be checked if in-memory one returns undefined
  asyncCache: new RedisCache<string>(ioRedis, {
    json: true, // this instructs loader to serialize passed objects as string and deserialize them back to objects
    ttlInMsecs: 1000 * 60 * 10,
  }),

  // data source will be generated from one or both provided data loading functions
  dataSourceGetOneFn: async (key: string) => {
    const results = await this.db('classifiers')
      .select('*')
      .where({
        id: parseInt(key),
      })
    return results[0]
  },
  dataSourceGetManyFn: (keys: string[]) => {
    return this.db('classifiers').select('*').whereIn('id', keys.map(parseInt))
  },
})

// If cache is empty, but there is data in the DB, after this operation is completed, both caches will be populated
const classifier = await loader.get('1')

Loader API

Loader has the following config parameters:

• throwIfUnresolved: boolean - if true, error will be thrown if all data sources return undefined;
• throwIfLoadError: boolean - if true, error will be thrown if any Loader throws an error;
• cacheUpdateErrorHandler: LoaderErrorHandler - error handler to use when cache throws an error during update;
• loadErrorHandler: LoaderErrorHandler - error handler to use when non-last data source throws an error during data retrieval.
• cacheKeyFromLoadParamsResolver: CacheKeyResolver<LoadParams> - mapper from LoadParams to a cache key. Defaults to a simple string passthrough when LoadParams are just a string key to begin with (which is the default)
• cacheKeyFromValueResolver: CacheKeyResolver<LoadParams> - mapper from entity to be cached to a cache key. Defaults to a dummy resolver which throws an error when methods that depend on it are used. Make sure to provide a real resolver if you are using the bulk API (getMany/getManyFromGroup)
• isEntryStillCurrentFn - optional lightweight staleness check that lets an entry entering the refresh window bump its TTL instead of refetching. See Conditional refresh with a staleness check.
• scheduleBackgroundWork: BackgroundWorkScheduler - optional hook for work the loader starts and does not await (background refreshes, staleness probes, notification publishes). Defaults to leaving that work detached. See Background work.

Loader provides following methods:

• invalidateCacheFor(key: string): Promise<void> - expunge all entries for given key from all caches of this Loader;
• invalidateCacheForMany(keys: string[]): Promise<void> - expunge all entries for given keys from all caches of this Loader;
• invalidateCache(): Promise<void> - expunge all entries from all caches of this Loader;
• applyRemoteInvalidationFor(key: string): void, applyRemoteInvalidationForMany(keys: string[]): void, applyRemoteValue(key: string, value: T | null): void, applyRemoteInvalidation(): void - apply an invalidation (or a value) that originated on another node, without re-publishing it. See Applying invalidations from your own transport;
• get(loadParams: LoadParams = string): Promise<T> - sequentially attempt to retrieve data for specified key from all caches and loaders, in an order in which those data sources passed to the Loader constructor.
• getMany(keys: string[], loadManyParams?: LoadManyParams = LoadParams): Promise<T> - sequentially attempt to retrieve data for specified keys from all caches and data sources, in an order in which those data sources were passed to the Loader constructor. Duplicate keys in the input array are automatically deduplicated to optimize performance and prevent redundant data source calls. Note that this retrieval mode doesn't support preemptive background refresh. Note that you need to manually resolve all keys upfront for this retrieval method (e. g. by using cacheKeyFromLoadParamsResolver from the Loader).

Parametrized loading

Sometimes you need to pass additional parameters for loader in case it will need to refill the cache, such as JWT token (for external calls) or additional query parameters (for a DB call). You can use optional generic LoadParams for that:

import type { DataSource } from 'layered-loader'

export type MyLoaderParams = {
  jwtToken: string
  entityId: string
}

export type MyLoaderManyParams = {
  jwtToken: string
}

class MyParametrizedDataSource implements DataSource<string, MyLoaderParams, MyLoaderManyParams> {
  async get(params: MyLoaderParams): Promise<string | undefined | null> {
    const resolvedValue = await someResolutionLogic(params.entityId, params.jwtToken)
    return resolvedValue
  }

  async getMany(entityIds: string[], params?: MyLoaderManyParams): Promise<string>[] {
    if (!params) {
        throw new Error('Load params are mandatory for MyParametrizedDataSource')
    } 
      
    const resolvedValues = await someBulkResolutionLogic(entityIds, params.jwtToken)
    return resolvedValues
  }
}

const loader = new Loader<string, MyLoaderParams>({
  inMemoryCache: IN_MEMORY_CACHE_CONFIG,
  dataSources: [new MyParametrizedDataSource()],
  cacheKeyFromLoadParamsResolver: (params) => params.entityId // if unique id consists of more than one field, you can concatenate them here
})
await operation.get({ jwtToken: 'someTokenValue', entityId: 'key' })

Update notifications

It is possible to mostly rely on fast in-memory caches and still keep data in sync across multiple nodes in a distributed system. In order to achieve this, you need to use Notification Publisher/Consumer pair. The way it works - whenever there is an invalidation event within the loader (invalidate, invalidateFor or invalidatForGroup methods are invoked), publisher sends a fanout notification to all subscribed consumers, and they invalidate their own caches as well.

Available notification adapters

| Transport | Package | One-liner | | --- | --- | --- | | Redis pub/sub (default) | Built-in (createNotificationPair from layered-loader) | Lowest latency, no per-instance queue setup, no lifecycle management. | | AWS SNS + SQS | @layered-loader/sqs | AWS-native fanout via one SNS topic and per-instance SQS queues. |

Both adapters implement the same notificationPublisher / notificationConsumer contract — the rest of your Loader configuration does not change when you swap one for the other.

Picking a notification adapter

Prefer Redis pub/sub. It is the simplest path operationally — no per-instance resources to provision or reap, no AWS quotas to worry about, no extra latency. Use Redis pub/sub whenever your stack already runs Redis (which it almost always does if you are using RedisCache).

The SNS/SQS adapter exists for two situations:

1. You cannot run Redis (e.g. a hard "AWS-managed services only" policy). Use the SNS/SQS adapter end-to-end. This works, but the per-instance SQS queue model has real operational costs — every restart with a fresh HOSTNAME leaks an SQS queue + SNS subscription unless you also use stable queue names.
2. You need to consume upstream AWS events (an SNS topic owned by another service) and turn them into cache invalidations. In this case the recommended pattern is a hybrid: Redis pub/sub for the cache cluster's own fanout, plus an SQS trigger reading the upstream topic. The trigger applies invalidations directly to your Loader, and the loader's Redis publisher handles fan-out. See Flexible invalidation triggers.

Redis pub/sub

Here is an example:

import Redis from 'ioredis'
import type { RedisOptions } from 'ioredis'
import { createNotificationPair, Loader } from 'layered-loader'

const redisOptions: RedisOptions = {
  host: 'localhost',
  port: 6379,
  password: 'sOmE_sEcUrE_pAsS',
}

export type User = {
  // some type
}

const redisPublisher = new Redis(redisOptions)
const redisConsumer = new Redis(redisOptions)
const redisCache = new Redis(redisOptions)

const { publisher: notificationPublisher, consumer: notificationConsumer } = createNotificationPair<User>({
  channel: 'user-cache-notifications',
  consumerRedis: redisConsumer, // you can pass redis config instead
  publisherRedis: redisPublisher, // you can pass redis config instead
})

const userLoader = new Loader({
  inMemoryCache: { ttlInMsecs: 1000 * 60 * 5 },
  asyncCache: new RedisCache<User>(redisCache, {
    ttlInMsecs: 1000 * 60 * 60,
  }),
  notificationConsumer,
  notificationPublisher,
})

await userLoader.init() // this will ensure that consumers have definitely finished registering on startup, but is not required

await userLoader.invalidateCacheFor('key') // this will transparently invalidate cache across all instances of your application

There is an equivalent for group loaders as well:

import Redis from 'ioredis'
import type { RedisOptions } from 'ioredis'
import { createGroupNotificationPair, GroupLoader } from 'layered-loader'

const redisOptions: RedisOptions = {
  host: 'localhost',
  port: 6379,
  password: 'sOmE_sEcUrE_pAsS',
}

export type User = {
  // some type
}

const redisPublisher = new Redis(redisOptions)
const redisConsumer = new Redis(redisOptions)
const redisCache = new Redis(redisOptions)

const { publisher: notificationPublisher, consumer: notificationConsumer } = createGroupNotificationPair<User>({
  channel: 'user-cache-notifications',
  consumerRedis: redisConsumer,
  publisherRedis: redisPublisher,
})

const userLoader = new GroupLoader({
  inMemoryCache: { ttlInMsecs: 1000 * 60 * 5 },
  asyncCache: new RedisCache<User>(redisCache, {
    ttlInMsecs: 1000 * 60 * 60,
  }),
  notificationConsumer,
  notificationPublisher,
})

await userLoader.init() // this will ensure that consumers have definitely finished registering on startup, but is not required

await userLoader.invalidateCacheFor('key', 'group') // this will transparently invalidate cache across all instances of your application

AWS SNS/SQS

@layered-loader/sqs provides a drop-in publisher/consumer pair backed by an SNS topic with one SQS queue per instance. The shape of the configuration mirrors the Redis pair — only the adapter changes.

How fanout works

SQS on its own is a competing-consumer queue: if every node read from the same queue, each invalidation would be delivered to only one of them and the rest would silently keep stale data. To get pub/sub-style fanout the adapter uses the SNS-fanout-to-SQS pattern:

1. There is one shared SNS topic (named in creationConfig.topic.Name — the same on every instance).
2. Each instance creates its own SQS queue subscribed to that topic. SNS delivers a copy of every published message to every subscribed queue.
3. Each instance consumes only its own queue, so it sees every invalidation exactly once.

This means each instance must pass a unique QueueName in its consumer's creationConfig.queue. The example below uses process.env.HOSTNAME for that — any per-instance identifier works (pod name, ECS task id, etc.). If two instances share a queue name they will share the queue and compete for messages, and roughly half the invalidations will be missed by each of them.

In locatorConfig mode the same rule applies, just shifted to provisioning: each instance must be pointed at its own pre-created queueUrl / subscriptionArn.

The publisher side is the opposite — all instances publish to the same topic ARN, so a single shared topic.Name in the publisher's creationConfig is correct (and required for the fanout to reach every subscriber).

import { Loader } from 'layered-loader'
import { createNotificationPair } from '@layered-loader/sqs'

const { publisher: notificationPublisher, consumer: notificationConsumer } =
  createNotificationPair<User>({
    publisher: {
      dependencies: pubDeps,
      creationConfig: { topic: { Name: 'user-cache-invalidations' } },
    },
    consumer: {
      dependencies: consumerDeps,
      creationConfig: {
        topic: { Name: 'user-cache-invalidations' },
        // Each instance MUST use a unique queue name (e.g. include the host id):
        queue: { QueueName: `user-cache-invalidations-${process.env.HOSTNAME}` },
      },
    },
  })

const userLoader = new Loader<User>({
  inMemoryCache: { ttlInMsecs: 1000 * 60 * 5 },
  asyncCache: yourAsyncCache,
  notificationConsumer,
  notificationPublisher,
})

await userLoader.invalidateCacheFor('key')

createGroupNotificationPair from the same package is the GroupLoader equivalent. See the package README for the full configuration reference (locator vs creation config, self-message filtering, AWS SDK dependencies, etc.).

Flexible invalidation triggers

Cache invalidations often need to fire in response to upstream domain events — a user.updated message published by another service onto an SNS topic, SQS queue, RabbitMQ exchange, or Kafka topic — rather than from writes inside the application that owns the cache. Wiring those events in usually means writing bespoke glue per service.

Layered-loader ships transport-agnostic primitives for that translation step, complementary to the cluster fanout described above:

• InvalidationAction / GroupInvalidationAction — the invalidation operations a resolver may emit.
• InvalidationResolver<TMessage, TAction> — a pure function (message) => action | action[] | null that maps an upstream message to invalidations.
• InvalidationTrigger — start() / stop() lifecycle interface implemented by every adapter.

A trigger consumes from one or more upstream sources, runs your resolver, and applies the resulting actions directly to a target — typically your Loader or GroupLoader. The loader already knows how to invalidate its own in-memory and async caches and (if you configured a notification pair) how to fan the invalidation out to peer instances, so the trigger reuses that machinery rather than introducing a second publication path.

Recommended pattern: Redis fanout + SQS trigger

If your upstream is AWS-native (an SNS topic owned by another service) but your own infrastructure runs Redis, this is the recommended setup. You get AWS-native event ingestion without the per-instance SQS queue lifecycle problem, because the trigger queue can be shared across every instance:

• The cache cluster's own notification pair is Redis pub/sub. Every Loader invalidation fans out via Redis — sub-ms latency, no per-instance queues, no cleanup.
• A single SQS queue is subscribed to the upstream SNS topic. Each instance runs the trigger code; SQS's competing-consumer semantics mean only one instance processes each upstream event.
• That instance's trigger calls loader.invalidateCacheFor(...), the loader writes through to the local in-memory + async caches, and the loader's Redis publisher propagates to every other instance.

import { z } from 'zod'
import Redis from 'ioredis'
import { createNotificationPair, Loader } from 'layered-loader'
import { SnsTopicInvalidationTrigger } from '@layered-loader/sqs'

const USER_EVENT = z.object({ type: z.literal('user.updated'), userId: z.string() })
const redisOptions = { host: 'redis', port: 6379 }

// 1. The cache cluster's own invalidation pair — pure Redis pub/sub.
const { publisher: notificationPublisher, consumer: notificationConsumer } =
  createNotificationPair<User>({
    channel: 'user-cache-invalidations',
    publisherRedis: new Redis(redisOptions),
    consumerRedis: new Redis(redisOptions),
  })

const userLoader = new Loader<User>({
  inMemoryCache: { ttlInMsecs: 1000 * 60 * 5 },
  asyncCache: yourAsyncCache,
  notificationConsumer,
  notificationPublisher,
})
await userLoader.init()

// 2. The trigger applies invalidations directly to the loader. The trigger
//    queue is SHARED across every instance (no ${HOSTNAME} suffix) — SQS
//    delivers each upstream event to exactly one of them.
const trigger = new SnsTopicInvalidationTrigger({
  target: userLoader,
  dependencies: consumerDeps,
  sources: [
    {
      creationConfig: {
        topic: { Name: 'domain-events.users' },                  // upstream service's topic
        queue: { QueueName: 'user-cache-invalidation-trigger' }, // SHARED across all instances
      },
      bindings: [
        { messageSchema: USER_EVENT, resolver: (msg) => ({ kind: 'delete', key: msg.userId }) },
      ],
    },
  ],
})
await trigger.start()

Why this is the right default for AWS-upstream consumption:

• No queue churn. One SQS queue exists, regardless of how many pods run.
• No lifecycle plumbing. No deleteQueueOnClose, no heartbeat, no reaper.
• Failure isolation. If a pod dies mid-message, SQS visibility timeout returns it to the queue and another pod picks it up.
• Cluster-wide fanout via Redis — every instance's in-memory cache reacts within milliseconds.

If your upstream events have meaningful per-entity ordering, use an SQS FIFO queue (QueueName: 'user-cache-invalidation-trigger.fifo'). Otherwise a standard queue is appropriate.

SNS/SQS adapter

The concrete adapter ships in @layered-loader/sqs as four trigger classes:

| Class | Source kind | Target | | --- | --- | --- | | SnsTopicInvalidationTrigger | Upstream SNS topic | flat Loader | | SqsQueueInvalidationTrigger | Existing SQS queue | flat Loader | | SnsTopicGroupInvalidationTrigger | Upstream SNS topic | GroupLoader | | SqsQueueGroupInvalidationTrigger | Existing SQS queue | GroupLoader |

Each trigger takes a single dependencies block and an array of sources (queues or topics) — listening to multiple at once is just listing them. Within one source, multiple event types can be routed to different resolvers via a messageTypeField discriminator. composeTriggers(...) bundles multiple triggers for deployments that need to mix source kinds.

Future adapters (RabbitMQ, Kafka, Google Pub/Sub, ...) reuse the same InvalidationAction / resolver primitives — only the consumer wiring changes. See the package README for the full reference, including group triggers, multi-source / multi-binding configuration, mixing source kinds via composeTriggers, and error handling.

Applying invalidations from your own transport

Applying an invalidation that arrived from somewhere else is not the same operation as originating one, and the library has separate methods for the two:

| | originating (invalidateCacheFor, invalidateCacheForGroup, ...) | applying (applyRemoteInvalidationFor, ...) | | --- | --- | --- | | in-memory cache | deletes | deletes | | running loads | fences | fences | | background refreshes | fences | fences | | async cache (Redis) | deletes | leaves alone — the origin already deleted it from the shared tier | | notification publisher | publishes | publishes nothing — the origin already broadcast it | | return type | Promise<void> (awaits the async tier) | void (purely local, synchronous) |

Do not reach for invalidateCacheFor to apply an invalidation you received. It re-publishes, so on a node that also has a publisher configured it echoes the invalidation straight back onto the bus.

The applyRemote* methods exist for transports the library does not ship, and in particular for pull-based ones: a host that reads "what changed since cursor N" at the start of a request and applies the result, because the runtime cannot hold a subscription open (see Isolate runtimes with no invalidation bus).

// at the start of a request, on a runtime where nothing can hold a subscription
const { keys, cursor } = await readInvalidationsSince(lastCursor)
for (const key of keys) {
  loader.applyRemoteInvalidationFor(key)
}
lastCursor = cursor

Flat caches (Loader, ManualCache):

• applyRemoteInvalidationFor(key)
• applyRemoteInvalidationForMany(keys)
• applyRemoteValue(key, value) — a value that was set on another node and broadcast here
• applyRemoteInvalidation() — a cache-wide clear

Group caches (GroupLoader, ManualGroupCache):

• applyRemoteInvalidationFor(key, group)
• applyRemoteInvalidationForGroup(group)
• applyRemoteValue(key, value, group)
• applyRemoteInvalidation()

All of them fence the affected key, so nothing that was already in flight when the invalidation arrived can write its pre-invalidation snapshot back into the in-memory cache when it resolves. That covers both a running load and a preemptive background refresh. A caller that was already awaiting that load still receives its value — exactly as for a locally originated invalidation.

If you implement AbstractNotificationConsumer yourself you get this behaviour for free: the target cache handed to setTargetCache is a facade over the owning loader that routes deletions and sets through these methods, so this.targetCache.delete(key) fences running loads too.

One thing the fence does not — and cannot — cover: the shared async tier. A background refresh that has already read from your data sources still writes the value it read into Redis, even if an invalidation arrives in between. That is the ordinary read-through race that any cache in front of a mutable store has (a plain cache miss has it too), and closing it needs versioning at the store, not a local fence. The fence is about this instance's in-memory tier: an invalidated entry stays invalidated locally instead of being resurrected, and the async tier converges on its own TTL.

Background work

The loader starts some work it does not await: preemptive background refreshes, staleness probes, and notification publishes. By default those promises are simply left detached, which is the right behaviour on Node.

scheduleBackgroundWork hands each of them to you instead:

const loader = new Loader<User>({
  inMemoryCache: { ttlInMsecs: 60_000, ttlLeftBeforeRefreshInMsecs: 20_000, cacheId: 'users' },
  dataSources: [userDataSource],

  scheduleBackgroundWork: (work, meta) => {
    // meta is { cacheId: 'users', reason: 'refresh' | 'notification' }
    pending.add(work)
    void work.finally(() => pending.delete(work))
  },
})

meta.reason is a closed set ('refresh' | 'notification') rather than a free-form string, because hosts branch on it in practice. meta.cacheId is the cacheId configured on inMemoryCache, so work from several loaders can be told apart in logs.

The promise you are given is guaranteed to settle fulfilled. The loader has already routed any error to loadErrorHandler / cacheUpdateErrorHandler / the publisher's error handler and attached its own handler before handing it over. That matters on runtimes where these promises get adopted by a request: ctx.waitUntil() on a rejecting promise fails the request that adopted it.

Three things this is useful for:

• Isolate runtimes, where a promise that outlives its request faults rather than merely leaking. See Hand background work to the request.
• Tests, which can await quiescence instead of sleeping and hoping.
• Graceful shutdown, which can drain in-flight refreshes before closing Redis connections.

Isolate runtimes with no invalidation bus

Cloudflare Workers — and by extension any target made of many short-lived, mutually unreachable instances (Deno Deploy, Vercel edge, Lambda@Edge, aggressively-scaled serverless Node) — break three assumptions that the notification pair is built on:

• there is no pub/sub any instance can reach;
• nothing can hold a subscription open between requests, so push is structurally unavailable and the answer has to be pull;
• I/O is scoped to the request that created it, so a promise that outlives its request faults with "Cannot perform I/O on behalf of a different request" unless it was handed to ctx.waitUntil().

layered-loader/core imports cleanly on those runtimes, and everything below is runtime-neutral — the library never learns what a Worker is. What follows is the deployment shape that works.

Hoist the loader to isolate scope

Construct the loader at module scope, once per isolate, not per request. A cache rebuilt on every invocation is a memo, not a cache — it can only ever serve the request that filled it.

This is also a correctness requirement once scheduleBackgroundWork is in play, not just a performance one: see below.

Pull invalidations instead of pushing them

notificationConsumer cannot be implemented where nothing can hold a subscription. Read what changed at the start of a request and apply it with applyRemoteInvalidationFor and friends. That is a supported use of the public API, and the applyRemote* methods exist precisely for it.

For a scope-wide generation counter — one row, or one Durable Object read, per tenant — the whole mechanism is a counter comparison plus one applyRemoteInvalidationForGroup when it moved:

const generation = await readTenantGeneration(tenantId)   // once per request, memoised
if (generation !== lastSeenGeneration.get(tenantId)) {
  loader.applyRemoteInvalidationForGroup(tenantId)
  lastSeenGeneration.set(tenantId, generation)
}

This is O(1) reads per request regardless of how many keys the request touches, and it needs nothing from the library beyond the methods above. Reach for it before reaching for a per-entry probe.

Hand background work to the request

Supply scheduleBackgroundWork so detached promises get adopted by the request that spawned them:

import { AsyncLocalStorage } from 'node:async_hooks'   // available on workerd

const requestCtx = new AsyncLocalStorage<ExecutionContext>()

// module scope: one loader per isolate, NOT one per request
const loader = new Loader<User>({
  inMemoryCache: { ttlInMsecs: 60_000 },
  dataSources: [userDataSource],
  scheduleBackgroundWork: (work) => {
    const ctx = requestCtx.getStore()
    if (ctx) {
      ctx.waitUntil(work)
    } else {
      void work
    }
  },
})

export default {
  fetch: (request, env, ctx) => requestCtx.run(ctx, () => handle(request, env)),
}

The hook must resolve the context at call time, as above. Closing over a ctx at construction is wrong: the loader lives at isolate scope, so that ctx belongs to whichever request happened to build it, and using it from a later request reintroduces the exact fault the hook exists to avoid.

Validating on every read

If you have a cheap per-entry version token (a document version, a branch head sha, a JWKS key id), isEntryStillCurrentFn can validate against it on every read rather than only near expiry — set ttlLeftBeforeRefreshInMsecs equal to ttlInMsecs, which puts every entry permanently inside the refresh window:

const loader = new Loader<Doc>({
  inMemoryCache: { ttlInMsecs: 60_000, ttlLeftBeforeRefreshInMsecs: 60_000 },
  dataSources: [docDataSource],
  isEntryStillCurrentFn: async (cachedValue, loadParams) =>
    (await readDocVersion(loadParams)) === cachedValue!.version,
})

Note what this does and does not give you. The probe runs on every hit, but it does not block the read: the cached value is served immediately and the probe decides what the next read sees. This is stale-while-revalidate, not must-revalidate. It shrinks the staleness window from ttlInMsecs to roughly one read, which is enough for most content, and is not enough where serving one stale answer is itself the bug (an authorization decision, say). For those, either do not cache, or invalidate on the write path via a generation counter as above.

Two properties of the probe worth knowing before you build on it:

• Concurrent reads of the same key share a single probe — the loader deduplicates them the same way it deduplicates loads, so you do not need to memoise on your side.
• Reads of different keys do not. isEntryStillCurrentFn is per entry, so a token shared across many keys (one branch head sha covering many files) is read once per key. If that is your shape, a scope-wide generation counter is the better fit, for the reason given above.

What the library does not do

• Nothing runs on a timer. There is no eviction sweep and no refresh tick; in-memory entries expire lazily when they are read. An in-memory-only loader schedules no periodic work at all, so there is nothing here that behaves differently under workerd's timer restrictions.
• There is no blocking (must-revalidate) read mode. A cache hit is never delayed by a probe. Adding one is not simply an extra option: getInMemoryOnly is synchronous and cannot await anything, and getMany serves an all-hit batch straight from memory without entering the refresh path at all — so a "no hit is ever served unvalidated" guarantee would be quietly false for two of the three read methods. If you need that guarantee today, use a generation counter on the write path rather than a per-read probe.
• No async cache tier is provided for edge KV stores. That is deliberate rather than missing: Redis in this library is an invalidation bus, not a data tier, and a store whose writes propagate eventually would be a weaker coherence story than the primary database it is meant to protect.

If your deployment can reach a message bus but you would rather not run Redis, @layered-loader/sqs gives you cross-instance invalidation over SNS/SQS and imports from layered-loader/core.

Cache statistics

You can keep track of your in-memory cache usage is by using special cache type - lru-object-statistics:

import { HitStatisticsRecord, Loader } from 'layered-loader'

const record = new HitStatisticsRecord()
const operation = new Loader({
  inMemoryCache: {
    ttlInMsecs: 99999,
    cacheId: 'some cache',
    globalStatisticsRecord: record,
    cacheType: 'lru-object-statistics',
  },
})

operation.getInMemoryOnly('value')

expect(record.records).toEqual({
  'some cache': {
    '2023-05-20': {
      cacheSize: 100, // how many elements does cache currently have
      evictions: 5, // how many elements were evicted due to cache being at max capacity
      expirations: 0, // how many elements were removed during get due to their ttl being exceeded
      hits: 0, // how many times element was successfully retrieved from cache during get
      emptyHits: 0, // out of all hits, how many were null, undefined or ''?
      falsyHits: 0, // out of all hits, how many were falsy?
      misses: 1, // how many times element was not in cache or expired during get
      invalidateOne: 1, // how many times element was invalidated individually
      invalidateAll: 2, // how many times entire cache was invalidated
      sets: 0, // how many times new element was added
    },
  },
})

Note that statistics accumulation affects performance of the cache, and it is recommended to only enable it temporarily, while conducting cache effectiveness analysis.

Cache-only operations

Sometimes you may want to avoid implementing loader in the chain (e. g. when retrieval is too complex to be fit into a single key paradigm), while still having a sequence of caches. In that case you can define a caching operation:

const cache = new ManualCache<string>({
  // this cache will be checked first
  inMemoryCache: {
    ttlInMsecs: 1000 * 60,
    maxItems: 100,
  },

  // this cache will be checked if in-memory one returns undefined
  asyncCache: new RedisCache<string>(ioRedis, {
    json: true, // this instructs loader to serialize passed objects as string and deserialize them back to objects
    ttlInMsecs: 1000 * 60 * 10,
  }),
})

// this will populate all caches
await cache.set('1', 'someValue')

// If any of the caches are still populated at the moment of this operation, 'someValue' will propagate across all caches
const classifier = await cache.get('1')

Note that Loaders are generally recommended over ManualCaches, as they offer better performance: LoadingOperations deduplicate all the get requests that come during the window between checking the cache and populating it, while Caching Operation will resolve all of them to undefined after checking the cache, both increasing load on the cache, and also potentially invoking the loading logic multiple times.

Forcing an update

In certain cases you may want to fetch fresh data from the datasource before invalidating the cache. In that case you should use the forceRefresh method:

// This will resolve the latest version of the data for the key "1", update async and inmemory caches and fire a NotificationPublisher invalidation command, if publisher is set  
await cache.forceRefresh('1')

Forcing a specific value

In certain cases you may want to explicitly store a specific value in all of your caches layers. In that case you should use the forceSetValue method:

// This will set the value of all configured caches for the key "1" to a value "newValue", and fire a NotificationPublisher set value command, if publisher is set  
await cache.forceSetValue('1', 'newValue')

For a GroupLoader use forceSetValueForGroup, which takes the extra group argument. Note that group notification publishers only broadcast deletions, so no set command is published; other nodes converge via their own TTL expiry.

await groupCache.forceSetValueForGroup('1', 'newValue', 'group1')

Usage in high-performance systems

Synchronous short-circuit

In case you are handling very heavy load and want to achieve highest possible performance, you can avoid asynchronous retrieval (and unnecessary Promise overhead) altogether in case there is a value already available in in-memory cache. Here is the example:

const loader = new Loader<MyValueType>({
  inMemoryCache: {
    // configuration here
  },

  // this cache will be checked if in-memory one returns undefined
  asyncCache: new RedisCache<MyValueType>(ioRedis, {
    // configuration here
  }),
  dataSources: [new MyDataSource()],
})

const cachedValue =
  // this very quickly checks if we have value in-memory
  loader.getInMemoryOnly('key') ||
  // if we don't, proceed with checking asynchronous cache and datasources
  (await loader.getAsyncOnly('key'))

Note that this will only work with truthy values. If you expect to get significant amount of falsy values (null for non-existing entries or 0/false), you should use an extended short-circuit syntax:

let cachedValue: MyValueType | undefined | null
cachedValue = loader.getInMemoryOnly('key')

if (cachedValue === undefined) {
  cachedValue = await loader.getAsyncOnly('key')
}

If you are unsure, whether you are caching significant amount of falsy or empty (null/empty string) values, you can enable cache statistics for discovering this data. See section "Cache statistics" for how to set that up.

Preemptive background refresh

In case some of your datasource calls are very expensive, and you want to reduce response latency, you can start preemptively refreshing your cache in background while still serving not-yet-stale current data. In order to do so, you need to set parameter ttlLeftBeforeRefreshInMsecs. For in-memory cache:

const operation = new Loader<string>({
  inMemoryCache: {
    cacheId: 'some-cache',
    ttlInMsecs: 1000 * 60,
    ttlLeftBeforeRefreshInMsecs: 1000 * 20, // this means that when there is a GET operation for the cache entry, and it has less than 20 seconds of TTL left, background refresh for this entry will start
  },
  // the rest of loader configuration
})

For Redis cache:

const asyncCache = new RedisCache<string>(redis, {
  ttlInMsecs: 1000 * 60,
  ttlLeftBeforeRefreshInMsecs: 1000 * 20,
}) // this means that when there is a GET operation for the cache entry, and it has less than 20 seconds of TTL left, background refresh for this entry will start

Note that there is overhead involved in performing refresh checks (especially for Redis). Always measure performance before and after enabling preemptive refresh in order to determine, whether it improves or worsens the performance of your system. Bulk operations (getMany()) do not support preemptive background refresh — when every requested key hits in memory, the batch is served without entering the refresh path at all.

The refresh itself is fire-and-forget: nothing schedules it on a timer, and the read that triggers it is not delayed by it. If your runtime needs those promises tracked rather than detached, supply scheduleBackgroundWork.

Conditional refresh with a staleness check

If refetching an entry is expensive, but verifying that it is still up-to-date is cheap (e. g. comparing an updatedAt timestamp, a version column or a content hash), you can avoid unnecessary refetches by providing isEntryStillCurrentFn. This is the loader-level equivalent of HTTP conditional revalidation (ETag / If-Modified-Since).

When an entry enters the ttlLeftBeforeRefreshInMsecs window, the loader first invokes your check with the cached value instead of immediately starting a full background refresh:

• if it resolves to true, the entry's TTL is reset to the full ttlInMsecs, and no data source call is made;
• if it resolves to false (or throws, or the entry disappeared in the meantime), the usual full background refresh from the data sources runs.

The check never delays the read that triggered it. That read is served the cached value either way, and the check decides what the next read sees — this is stale-while-revalidate, not must-revalidate. There is no blocking read mode; see What the library does not do for why.

The check works both with an async cache and with an in-memory-only loader. It only needs a preemptive refresh window (ttlLeftBeforeRefreshInMsecs) to fire inside: configure it on the asyncCache, or, for a loader that has no async tier, on the inMemoryCache. When both tiers have a refresh window, the async cache takes precedence and the check runs on its refresh path.

const operation = new Loader<UserEntity>({
  asyncCache: new RedisCache<UserEntity>(redis, {
    json: true,
    ttlInMsecs: 1000 * 60,
    ttlLeftBeforeRefreshInMsecs: 1000 * 20,
  }),
  dataSources: [expensiveUserDataSource],
  isEntryStillCurrentFn: async (cachedValue, loadParams) => {
    // light query instead of refetching the whole entity
    const updatedAt = await getUserUpdatedAt(loadParams)
    return updatedAt.getTime() === new Date(cachedValue!.updatedAt).getTime()
  },
})

For GroupLoader the check additionally receives the group:

const operation = new GroupLoader<UserEntity>({
  asyncCache: new RedisGroupCache<UserEntity>(redis, {
    json: true,
    ttlInMsecs: 1000 * 60,
    ttlLeftBeforeRefreshInMsecs: 1000 * 20,
  }),
  dataSources: [expensiveUserDataSource],
  isEntryStillCurrentFn: async (cachedValue, loadParams, group) => {
    const updatedAt = await getUserUpdatedAt(loadParams, group)
    return updatedAt.getTime() === new Date(cachedValue!.updatedAt).getTime()
  },
})

For a loader with no async cache, configure the refresh window on the in-memory cache instead - the check then runs against the in-memory value:

const operation = new Loader<UserEntity>({
  inMemoryCache: {
    ttlInMsecs: 1000 * 60,
    ttlLeftBeforeRefreshInMsecs: 1000 * 20,
  },
  dataSources: [expensiveUserDataSource],
  isEntryStillCurrentFn: async (cachedValue, loadParams) => {
    // light query instead of refetching the whole entity
    const updatedAt = await getUserUpdatedAt(loadParams)
    return updatedAt.getTime() === new Date(cachedValue!.updatedAt).getTime()
  },
})

Things to keep in mind:

• isEntryStillCurrentFn requires a preemptive refresh window (ttlLeftBeforeRefreshInMsecs) to run inside, on either the asyncCache or the inMemoryCache. When it runs on the async path, the async cache must implement resetTtl (resetTtlFromGroup for group caches); RedisCache and RedisGroupCache implement it, and if you implement the Cache int