[Cache] add integration with Messenger to allow computing cached values in a worker

[nicolas-grekas]

Q	A
Branch?	master
Bug fix?	no
New feature?	yes
BC breaks?	no
Deprecations?	no
Tests pass?	yes
Fixed tickets	-
License	MIT
Doc PR	-

This PR needs and for now embeds #30334. See 2nd commit.

Using the new CacheInterface enables probabilistic early expiration by default. This means that from time to time, items are elected as early-expired while they are still fresh (see Wikipedia for details).

This PR adds a new early_expiration_message_bus option when configuring cache pools. When this option is set, cache pools are configured to send those early-expired items on a Messenger bus, then serve the current value immediately, while the updated value is computed in a worker in the background.

CacheInterface::get($key, $callback) accepts any callable, but sending any callable on a bus is not possible (e.g. a Closure cannot be serialized). To bypass this constraint, this feature works only with callables in the form [$service, 'publicMethod'], where $service is any public or reversible service.

This constraint is a serious one: this $service must compute a value when knowing only its key. This means keys should embed enough information for this to happen. I think that's not that hard - and we may find ways to provide additional context in the future.

At least the goal is worth it: in theory, this strategy allows achieving a 100% hit ratio even when invalidation-by-expiration is used.

There are two things one needs to do to enable this behavior:

1. bind a message bus to a cache pool:

framework:
    cache:
        pools:
            test.cache:
                early_expiration_message_bus: messenger.default_bus

2. route EarlyExpirationMessage to a transport:

framework:
    messenger:
        routing:
            'Symfony\Component\Cache\Messenger\EarlyExpirationMessage': amqp

[Nyholm]

Cool. I really like this feature.

Could you show a code example how I would use it.

I’m also trying to figure out how I can work around the limitation..

[nicolas-grekas]

@Nyholm thanks for having a look :)

To test this, you need to configure a pool with the messenger_bus option set, and have the consumer run:

framework:
    cache:
        pools:
            test.cache:
                messenger_bus: message_bus
    messenger:
        transports:
            amqp: '%env(MESSENGER_TRANSPORT_DSN)%'
        routing:
            'Symfony\Component\Cache\Messenger\Message': amqp

BTW, this shows another area where Messenger needs to improve: how should a bundle provide configuration for this Message-to-transport mapping out of the box? ping @sroze @ogizanagi @weaverryan.

Then e.g. in a controller:

    /**
     * @Route("/test", name="test")
     */
    public function index(CacheInterface $testCache)
    {
        $value = $testCache->get('foo', [$this, 'compute']);

        return $this->render('test/index.html.twig', [
            'value' => $value,
        ]);
    }

    public function compute(ItemInterface $item)
    {
        $item->expiresAfter(5);
        sleep(1);

        return sprintf('#%06X', mt_rand(0, 0xFFFFFF));
    }

Runnig the dev weberver: symfony serve
Running the worker: bin/console messenger:consume -vvv
etc.

[nicolas-grekas]

See https://github.com/nicolas-grekas/cache-bus for a running example.

[sroze]

Looks like a great idea 😍

[tgalopin]

What's the status of this PR? I feel everyone concerned is at the EU hackathon, could be a great opportunity to merge :)

[sroze]

It's black magic

[fabpot]

Can you rebase and perhaps answers the few questions?

[Plopix]

I would love to see this merged/updated to be merged! I think that is a great addition for performances

[nicolas-grekas]

I'm adding tests, that should be ok in a few hours.

[nicolas-grekas]

I rebased the PR and added unit tests. It works \o/

I also verified on the demo app I built last year, all works as expected 🎉

Status: needs review

[bastnic]

Stunning idea!

[fabpot]

@nicolas-grekas Can you fix the tests?

[fabpot]

Thank you @nicolas-grekas.