diff --git a/README.md b/README.md index ea45878..b8e1b91 100644 --- a/README.md +++ b/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..`). 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. diff --git a/config/laya.php b/config/laya.php index b04ac61..d3bcdf7 100644 --- a/config/laya.php +++ b/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. diff --git a/src/Laravel/LayaServiceProvider.php b/src/Laravel/LayaServiceProvider.php index 12faeb2..24bab77 100644 --- a/src/Laravel/LayaServiceProvider.php +++ b/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), diff --git a/src/Laya.php b/src/Laya.php index f6804c7..3233b5b 100644 --- a/src/Laya.php +++ b/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,6 +62,8 @@ 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 @@ -67,6 +71,7 @@ final class Laya * @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..', 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 diff --git a/tests/CacheTest.php b/tests/CacheTest.php index 765848a..7f536ca 100644 --- a/tests/CacheTest.php +++ b/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 */ public array $items = []; + /** @var array */ + 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|int $health + * @param list $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 $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 $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); +}); diff --git a/tests/LaravelTest.php b/tests/LaravelTest.php index 289664a..d6f6bdb 100644 --- a/tests/LaravelTest.php +++ b/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);