Add versionCache to key cached predictions by checkpoint revisions

README.md

@@ -57,6 +57,7 @@ LAYA_TIMEOUT=       # seconds; needs symfony/http-client (see below)
 LAYA_RETRIES=       # retries on 503 and transport errors, with backoff
 LAYA_CACHE_STORE=   # e.g. redis, to cache predictions (see Caching)
 LAYA_CACHE_TTL=     # seconds
+LAYA_CACHE_VERSIONED=false        # add the checkpoint revisions to the cache keys (see Caching)
 LAYA_EVENTS=true                  # dispatch PredictionMade/PredictionFailed/CacheFailed (see Events)
 LAYA_EVENTS_INCLUDE_STATE=false
 ```
@@ -588,7 +589,13 @@ Laya gives the same answer to the same input, so you can skip repeat requests wi
 $laya = new Laya('http://laya:8000', cache: $psr16Cache, cacheTtl: 86400);
 ```
 
-The key doesn't include the checkpoint revision, so clear the cache (or set a TTL) when you upgrade `laya-serve`'s checkpoints.
+By default the key doesn't include the checkpoint revision, so clear the cache (or set a TTL) when you upgrade `laya-serve`'s checkpoints. Or let Laya do it:
+
+```php
+$laya = new Laya('http://laya:8000', cache: $psr16Cache, cacheTtl: 86400, versionCache: true);
+```
+
+With `versionCache`, the first lookup per instance calls `/health` once and puts the checkpoint revisions into the key (`laya.<revisionhash>.<bodyhash>`). The hash is also kept in the cache for 60 seconds, so short-lived PHP-FPM requests don't each call `/health`. A changed revision means new keys; the old entries are never hit again and expire through the TTL. If `/health` fails or reports no revisions (sys1), Laya uses the plain key and the prediction goes ahead. It's off by default, since turning it on changes every key. In Laravel, set `LAYA_CACHE_VERSIONED=true`.
 
 The cache is best-effort: it never fails a prediction. If reading it throws (say, Redis is unreachable), or an entry isn't a laya response, Laya asks `laya-serve` and overwrites the entry with the answer. If writing it throws, you get the result anyway, uncached; in a batch, every state still gets its result. Only exceptions are caught, so an `Error` still surfaces. With an event dispatcher, each failure dispatches a `CacheFailed` (see Events); without one, failures are silent.
 

config/laya.php

@@ -23,10 +23,11 @@
     'batch_size' => (int) env('LAYA_BATCH_SIZE', 64),
 
     // A cache store (e.g. "redis") to remember predictions in; null disables caching.
-    // Clear it, or set a TTL, when you upgrade laya-serve's checkpoints.
+    // Clear it, set a TTL, or turn on versioned when you upgrade laya-serve's checkpoints.
     'cache' => [
         'store' => env('LAYA_CACHE_STORE'),
         'ttl' => env('LAYA_CACHE_TTL'), // seconds; null keeps entries as long as the store does
+        'versioned' => env('LAYA_CACHE_VERSIONED', false), // add laya-serve's checkpoint revisions to the keys; costs a /health request
     ],
 
     // Dispatch PredictionMade, PredictionFailed and CacheFailed through the app's event dispatcher.

src/Laravel/LayaServiceProvider.php

@@ -35,6 +35,7 @@ public function register(): void
                 apiKey: is_string($apiKey) && $apiKey !== '' ? $apiKey : null,
                 cache: is_string($store) && $store !== '' ? $this->app->make(CacheManager::class)->store($store) : null,
                 cacheTtl: is_numeric($ttl) ? (int) $ttl : null,
+                versionCache: filter_var($config->get('laya.cache.versioned'), FILTER_VALIDATE_BOOLEAN),
                 events: filter_var($config->get('laya.events.enabled'), FILTER_VALIDATE_BOOLEAN) ? new EventDispatcher($this->app) : null,
                 includeState: filter_var($config->get('laya.events.include_state'), FILTER_VALIDATE_BOOLEAN),
                 httpClient: $this->httpClient($config),

src/Laya.php

@@ -46,6 +46,8 @@ final class Laya
      */
     public const string AUTO_MODEL = 'jev-latest';
 
+    private const int REVISION_TTL = 60; // @pest-mutate-ignore
+
     /** The keys of a minConfidence map: core's temp_bucket spelling, type and option count, or "default". */
     private const string BUCKET = '/^(?:(?:choice|score|noul):(?:2|3-5|6-10|11\+)|default)$/D';
 
@@ -60,13 +62,16 @@ final class Laya
 
     private readonly StreamFactoryInterface $streamFactory;
 
+    private ?string $keyPrefix = null;
+
     /**
      * @param  ClientInterface|null  $httpClient  any PSR-18 client; discovered when omitted
      * @param  CacheInterface|null  $cache  caches predictions by state, questions and model; laya is deterministic. Best-effort: its exceptions never fail a prediction
      * @param  int|\DateInterval|null  $cacheTtl  null keeps entries as long as the cache does
      * @param  EventDispatcherInterface|null  $events  receives a PredictionMade or PredictionFailed for every prediction, and a CacheFailed when the cache fails
      * @param  bool  $includeState  put the state on the events; off by default, since states may be sensitive
      * @param  int  $batchSize  states per batch request; laya-serve refuses more than its MAX_BATCH_STATES (64 by default)
+     * @param  bool  $versionCache  add laya-serve's checkpoint revisions to the cache keys, so a checkpoint upgrade stops serving old answers. Costs one /health request per instance (or per minute, through the cache); off by default, and changes every key
      */
     public function __construct(
         string $baseUrl = 'http://localhost:8000',
@@ -79,6 +84,7 @@ public function __construct(
         private readonly ?EventDispatcherInterface $events = null,
         private readonly bool $includeState = false,
         int $batchSize = 64,
+        private readonly bool $versionCache = false,
     ) {
         $batchSize >= 1 || throw new InvalidOptionException(sprintf('batchSize must be at least 1, got %d.', $batchSize));
         $this->batchSize = $batchSize;
@@ -178,7 +184,7 @@ public function predict(string|array|\JsonSerializable $state, array $questions,
     {
         self::assertMinConfidence($minConfidence);
         $json = self::json($this->body(['state' => $state], self::wire($questions), $model, $maxLen, $headMaxLen, $minConfidence));
-        $key = self::cacheKey($json);
+        $key = $this->cacheKey($json);
         $cached = $this->cached($key);
         if ($cached !== null) {
             $this->events?->dispatch($this->made($state, $questions, $model, $cached, true, 0.0));
@@ -228,7 +234,7 @@ public function predictMany(array $states, array $questions, ?Model $model = nul
         $results = [];
         $misses = [];
         foreach ($states as $id => $state) {
-            $key = self::cacheKey(self::json($this->body(['state' => $state], $wire, $model, $maxLen, $headMaxLen, $minConfidence)));
+            $key = $this->cacheKey(self::json($this->body(['state' => $state], $wire, $model, $maxLen, $headMaxLen, $minConfidence)));
             // Misses stay null, placeholders that keep the input order for the answers filled in below.
             $results[$id] = $this->cached($key);
             if ($results[$id] !== null) {
@@ -630,10 +636,53 @@ private static function json(array $body): string
         return json_encode($body, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
     }
 
-    // The key ignores the checkpoint revision; set a cacheTtl or clear the cache after upgrading laya.
-    private static function cacheKey(string $json): string
+    private function cacheKey(string $json): string
+    {
+        return $this->keyPrefix().hash('xxh128', $json);
+    }
+
+    // Without versionCache the key ignores the checkpoint revision; set a cacheTtl or clear the cache after upgrading laya.
+    private function keyPrefix(): string
+    {
+        if (! $this->versionCache || $this->cache === null) {
+            return 'laya.';
+        }
+
+        return $this->keyPrefix ??= $this->revisionPrefix($this->cache);
+    }
+
+    /**
+     * 'laya.<revisionhash>.', or the plain prefix when /health fails or reports no revisions. The hash lives in the
+     * cache for a minute, so short-lived processes don't each ask /health; a failed lookup isn't stored, the next process retries.
+     */
+    private function revisionPrefix(CacheInterface $cache): string
     {
-        return 'laya.'.hash('xxh128', $json);
+        $cacheKey = 'laya.revisions.'.hash('xxh128', $this->baseUrl);
+        try {
+            $hash = $cache->get($cacheKey);
+        } catch (\Exception $e) {
+            $this->cacheFailed('get', $cacheKey, $e);
+            $hash = null;
+        }
+        if (! is_string($hash)) {
+            try {
+                $revisions = $this->health()->revisions;
+            } catch (LayaException) {
+                return 'laya.';
+            }
+            if ($revisions === []) {
+                return 'laya.';
+            }
+            ksort($revisions);
+            $hash = hash('xxh128', self::json($revisions));
+            try {
+                $cache->set($cacheKey, $hash, self::REVISION_TTL);
+            } catch (\Exception $e) {
+                $this->cacheFailed('set', $cacheKey, $e);
+            }
+        }
+
+        return 'laya.'.$hash.'.';
     }
 
     private function post(string $path, string $json): RequestInterface

tests/CacheTest.php

@@ -2,10 +2,16 @@
 
 declare(strict_types=1);
 
+use GuzzleHttp\Psr7\Response;
 use MarcReichel\Laya\Events\CacheFailed;
 use MarcReichel\Laya\Events\PredictionMade;
 use MarcReichel\Laya\Exceptions\ServerException;
+use MarcReichel\Laya\Laya;
+use MarcReichel\Laya\Question;
 use Psr\EventDispatcher\EventDispatcherInterface;
+use Psr\Http\Client\ClientInterface;
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
 use Psr\SimpleCache\CacheInterface;
 
 /** A PSR-16 cache whose reads and writes throw $error when told to. */
@@ -14,6 +20,9 @@ final class FailingCache implements CacheInterface
     /** @var array<string, mixed> */
     public array $items = [];
 
+    /** @var array<string, null|int|DateInterval> */
+    public array $ttls = [];
+
     public function __construct(private ?Throwable $getError = null, private ?Throwable $setError = null) {}
 
     public function get(string $key, mixed $default = null): mixed
@@ -31,6 +40,7 @@ public function set(string $key, mixed $value, null|int|DateInterval $ttl = null
             throw $this->setError;
         }
         $this->items[$key] = $value;
+        $this->ttls[$key] = $ttl;
 
         return true;
     }
@@ -218,3 +228,131 @@ public function dispatch(object $event): object
     expect(fn () => layaBatching(cache: new FailingCache(getError: new TypeError('a bug')))->predict('x', questions()))->toThrow(TypeError::class, 'a bug')
         ->and(fn () => layaBatching(cache: new FailingCache(setError: new TypeError('a bug')))->predict('x', questions()))->toThrow(TypeError::class, 'a bug');
 });
+
+/**
+ * A Laya whose /health answers with $health (an array, or an int status for a failure) and whose predictions answer LAYA_RESPONSE.
+ *
+ * @param  array<string, mixed>|int  $health
+ * @param  list<RequestInterface>  $sent
+ */
+function layaVersioned(CacheInterface $cache, array|int $health, array &$sent = [], bool $versionCache = true, ?RecordingDispatcher $events = null, bool $noCache = false): Laya
+{
+    $client = new class($health, $sent) implements ClientInterface
+    {
+        /** @param list<RequestInterface> $sent */
+        public function __construct(private array|int $health, private array &$sent) {}
+
+        public function sendRequest(RequestInterface $request): ResponseInterface
+        {
+            $this->sent[] = $request;
+            if ($request->getMethod() === 'GET') {
+                return is_int($this->health) ? new Response($this->health, [], '{}') : new Response(200, [], json_encode($this->health));
+            }
+            $body = json_decode((string) $request->getBody(), true);
+            $results = array_map(fn ($state) => LAYA_RESPONSE, $body['states'] ?? [$body['state']]);
+
+            return new Response(200, [], json_encode(isset($body['states']) ? ['results' => $results] : $results[0]));
+        }
+    };
+
+    return new Laya('http://laya.local', httpClient: $client, cache: $noCache ? null : $cache, events: $events, versionCache: $versionCache);
+}
+
+function ask(Laya $laya): void
+{
+    $laya->predict('a ticket', ['q' => Question::choice('Q?', ['a', 'b'])]);
+}
+
+/** @param list<RequestInterface> $sent */
+function healthRequests(array $sent): int
+{
+    return count(array_filter($sent, fn (RequestInterface $r) => $r->getMethod() === 'GET'));
+}
+
+it('keeps the plain keys and never calls /health by default', function () {
+    $cache = new FailingCache;
+    $sent = [];
+    ask(layaVersioned($cache, ['status' => 'ok', 'revisions' => ['m' => 'abc']], $sent, versionCache: false));
+
+    expect(array_keys($cache->items))->toHaveCount(1)
+        ->and(array_keys($cache->items)[0])->toMatch('/^laya\.[0-9a-f]{32}$/')
+        ->and(healthRequests($sent))->toBe(0);
+});
+
+it('does not call /health without a cache', function () {
+    $sent = [];
+    $laya = layaVersioned(new FailingCache, ['status' => 'ok'], $sent, noCache: true);
+    ask($laya);
+
+    expect(healthRequests($sent))->toBe(0);
+});
+
+it('puts the revisions into the key, so changed revisions mean new keys and the same ones the same keys', function () {
+    $keys = [];
+    foreach ([['m' => 'abc', 'n' => null], ['n' => null, 'm' => 'abc'], ['m' => 'def', 'n' => null]] as $revisions) {
+        $cache = new FailingCache;
+        ask(layaVersioned($cache, ['status' => 'ok', 'revisions' => $revisions]));
+        $keys[] = array_keys($cache->items)[1]; // [0] is the cached revision hash
+    }
+
+    expect($keys[0])->toMatch('/^laya\.[0-9a-f]{32}\.[0-9a-f]{32}$/')
+        ->and($keys[1])->toBe($keys[0])
+        ->and($keys[2])->not->toBe($keys[0]);
+});
+
+it('asks /health once per instance, and shares the revision hash through the cache', function () {
+    $cache = new FailingCache;
+    $sent = [];
+    $laya = layaVersioned($cache, ['status' => 'ok', 'revisions' => ['m' => 'abc']], $sent);
+    ask($laya);
+    ask($laya);
+    $other = [];
+    ask(layaVersioned($cache, 503, $other));
+
+    expect(healthRequests($sent))->toBe(1)
+        ->and(healthRequests($other))->toBe(0)
+        ->and(array_keys($cache->items))->toHaveCount(2);
+});
+
+it('falls back to the plain key when /health fails or reports no revisions', function (array|int $health) {
+    $cache = new FailingCache;
+    $sent = [];
+    $laya = layaVersioned($cache, $health, $sent);
+    ask($laya);
+    ask($laya);
+
+    expect(array_keys($cache->items)[0])->toMatch('/^laya\.[0-9a-f]{32}$/')
+        ->and($cache->items)->toHaveCount(1)
+        ->and(healthRequests($sent))->toBe(1);
+})->with([
+    'health fails' => [503],
+    'sys1, no revisions' => [['status' => 'ok']],
+]);
+
+it('keeps the revision hash for a minute, under a key of its own', function () {
+    $cache = new FailingCache;
+    ask(layaVersioned($cache, ['status' => 'ok', 'revisions' => ['m' => 'abc']]));
+    $key = array_keys($cache->items)[0];
+
+    expect($key)->toMatch('/^laya\.revisions\.[0-9a-f]{32}$/')
+        ->and($cache->ttls[$key])->toBe(60);
+});
+
+it('versions the keys of predictMany() too', function () {
+    $cache = new FailingCache;
+    layaVersioned($cache, ['status' => 'ok', 'revisions' => ['m' => 'abc']])
+        ->predictMany(['a', 'b'], ['q' => Question::choice('Q?', ['a', 'b'])]);
+
+    expect(array_keys($cache->items))->toHaveCount(3)
+        ->and(array_keys($cache->items)[1])->toMatch('/^laya\.[0-9a-f]{32}\.[0-9a-f]{32}$/');
+});
+
+it('reports a cache that fails on the revision hash, and still predicts', function () {
+    $events = new RecordingDispatcher;
+    $sent = [];
+    $cache = new FailingCache(new RuntimeException('down'), new RuntimeException('down'));
+    ask(layaVersioned($cache, ['status' => 'ok', 'revisions' => ['m' => 'abc']], $sent, events: $events));
+
+    expect(array_map(fn (CacheFailed $e) => $e->operation, cacheFailures($events)))->toBe(['get', 'set', 'get', 'set'])
+        ->and(healthRequests($sent))->toBe(1);
+});

tests/LaravelTest.php

@@ -97,6 +97,17 @@
         ->and((fn () => $this->cacheTtl)->call($laya))->toBe(60);
 });
 
+it('versions the cache keys only when asked to', function () {
+    $versioned = fn () => (fn () => $this->versionCache)->call(app(Laya::class));
+
+    expect($versioned())->toBeFalse();
+
+    app()->forgetInstance(Laya::class);
+    config(['laya.cache.versioned' => 'true']);
+
+    expect($versioned())->toBeTrue();
+});
+
 it('does not cache by default, or with an empty store', function () {
     config(['laya.cache.store' => '']);
     $laya = app(Laya::class);