diff --git a/README.md b/README.md
index b86bc24..34a261c 100644
--- a/README.md
+++ b/README.md
@@ -21,7 +21,13 @@ Pure non-blocking (fiber based) strictly typed full-featured PHP driver for NATS
   - [Get message](#get-message)
 - [NATS KV](https://docs.nats.io/nats-concepts/jetstream/key-value-store)
   - [Store key values](#store-key-values)
+  - [Create and update buckets](#create-and-update-buckets)
+  - [Create, update and expire keys](#create-update-and-expire-keys)
+  - [Keys and history](#keys-and-history)
+  - [Bucket status](#bucket-status)
+  - [Purge deletes](#purge-deletes)
   - [Watch KV](#watch-kv)
+  - [Mirrors and sources](#mirrors-and-sources)
 - [NATS ObjectStore](https://docs.nats.io/nats-concepts/jetstream/obj_store)
   - [Store objects](#store-objects-in-the-buckets)
   - [Watch Object Store](#watch-object-store)
@@ -771,7 +777,10 @@ dump(
 $nc->drain();
 ```
 
-#### Watch KV
+#### Create and update buckets
+
+`createOrUpdateKeyValue()` creates the bucket or updates its config. `createKeyValue()` only creates it and fails with `StreamNameAlreadyInUse` if the bucket exists with another config,
+`updateKeyValue()` only updates it and fails with `StreamNotFound` if there is no bucket.
 
 ```php
 <?php
@@ -781,28 +790,344 @@ declare(strict_types=1);
 require __DIR__ . '/vendor/autoload.php';
 
 use Thesis\Nats;
+use Thesis\Nats\Exception\StreamNotFound;
 use Thesis\Nats\JetStream\KeyValue\BucketConfig;
-use function Amp\trapSignal;
 
 $nc = new Nats\Client(Nats\Config::default());
 $js = $nc->jetStream();
 
 $js->deleteKeyValue('configs');
 
+try {
+    $js->updateKeyValue(new BucketConfig(bucket: 'configs'));
+} catch (StreamNotFound) {
+    dump('There is no bucket to update yet.');
+}
+
+$js->createKeyValue(new BucketConfig(bucket: 'configs'));
+
+$kv = $js->updateKeyValue(new BucketConfig(bucket: 'configs', history: 5));
+
+dump($kv->status()->history); // 5
+
+$nc->drain();
+```
+
+#### Create, update and expire keys
+
+`create()` puts the value only if the key has no value: it was never put, or it was deleted or purged. `update()` puts the value only if the latest revision of the key is the given one.
+Both fail with `WrongLastSequence` otherwise. `put()`, `create()` and `update()` take a TTL of the entry, which requires the bucket to have `limitMarkerTtl`:
+once the entry expires, the server places a marker that `get()` returns as a purged entry, and the key can be created again.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\Exception\WrongLastSequence;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+use Thesis\Time\TimeSpan;
+use function Amp\delay;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
+$kv = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'locks',
+    limitMarkerTtl: TimeSpan::fromMinutes(1),
+));
+
+$revision = $kv->create('job.report', 'worker-1', ttl: TimeSpan::fromSeconds(10));
+
+try {
+    $kv->create('job.report', 'worker-2');
+} catch (WrongLastSequence) {
+    dump('The lock is taken.');
+}
+
+// Prolong the lock, only if nobody has changed it meanwhile.
+$revision = $kv->update('job.report', $revision, 'worker-1', ttl: TimeSpan::fromSeconds(10));
+
+// Release the lock: a deleted key can be created again.
+$kv->delete('job.report', $revision);
+$kv->create('job.report', 'worker-2');
+
+$kv->put('job.cleanup', 'worker-3', ttl: TimeSpan::fromSeconds(1));
+delay(2);
+dump($kv->get('job.cleanup')?->state); // EntryState::Purged
+
+$nc->drain();
+```
+
+#### Keys and history
+
+`keys()` returns the keys that have a value, optionally only the ones matching the filters, `history()` returns every stored revision of a key or of the keys matching a pattern, delete and purge markers included. Both read through a short-lived watch; the cancellation they take limits the whole read, the server requests included.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
 $kv = $js->createOrUpdateKeyValue(new BucketConfig(
     bucket: 'configs',
+    history: 10,
 ));
 
-$subscription = $kv->watch(static function (Nats\JetStream\KeyValue\Entry $entry): void {
-    dump("Config key {$entry->key} value changed to {$entry->value}");
-});
+$kv->put('app.env', 'prod');
+$kv->put('app.env', 'dev');
+$kv->put('app.debug', '1');
+$kv->put('database.dsn', 'mysql:host=127.0.0.1;port=3306');
+$kv->delete('app.debug');
+
+dump($kv->keys()); // ['app.env', 'database.dsn']
+dump($kv->keys('app.*')); // ['app.env']
+dump($kv->keys(['app.>', 'database.dsn'])); // ['app.env', 'database.dsn']
+
+foreach ($kv->history('app.env') as $entry) {
+    dump("{$entry->revision}: {$entry->value}");
+}
+
+$nc->drain();
+```
+
+#### Bucket status
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+use Thesis\Time\TimeSpan;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
+$kv = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'configs',
+    history: 5,
+    ttl: TimeSpan::fromHours(1),
+    compression: true,
+));
+
+$kv->put('app.env', 'prod');
+
+$status = $kv->status();
+
+dump(
+    $status->bucket,       // configs
+    $status->values,       // the number of stored revisions, markers included
+    $status->history,      // 5
+    $status->ttl?->toSeconds(), // 3600
+    $status->bytes,
+    $status->backingStore, // JetStream
+    $status->compressed,   // true
+);
+
+// The info of the stream behind the bucket.
+dump($status->info->config->maxBytes);
+
+$nc->drain();
+```
+
+#### Purge deletes
+
+`delete()` and `purge()` leave a marker, so that watches see the key is gone. `purgeDeletes()` removes the revisions of the deleted and purged keys and the markers older than the given age (30 minutes by default),
+the recent markers are kept. Pass zero to remove all markers.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+use Thesis\Time\TimeSpan;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
+$kv = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'configs',
+    history: 10,
+));
 
 $kv->put('app.env', 'prod');
+$kv->put('app.env', 'dev');
+$kv->delete('app.env');
+
+$kv->purgeDeletes();
+dump(\count($kv->history('app.env'))); // 1, the recent delete marker
+
+$kv->purgeDeletes(TimeSpan::fromSeconds(0));
+dump(\count($kv->history('app.env'))); // 0
+
+$nc->drain();
+```
+
+#### Watch KV
+
+`watch()` first calls the handler with the latest entry of every watched key, deleted keys included, and then with every change. It returns a `Watcher`:
+`awaitInitialValues()` returns once the handler has been called for every entry the bucket had when the watch started, so the watcher is up to date from then on.
+`stop()` stops the watch and deletes its consumer.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+use Thesis\Nats\JetStream\KeyValue\Entry;
+use function Amp\trapSignal;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
+$kv = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'configs',
+));
+
+$kv->put('app.env', 'prod');
+
+$watcher = $kv->watch(static function (Entry $entry): void {
+    dump("Config key {$entry->key} is {$entry->value}, revision {$entry->revision}");
+});
+
+$watcher->awaitInitialValues();
+dump('The current configs are loaded.');
+
+$kv->put('app.env', 'dev');
 $kv->put('database.dsn', 'mysql:host=127.0.0.1;port=3306');
 
 trapSignal([\SIGTERM, \SIGINT]);
 
-$subscription->stop();
+$watcher->stop();
+
+$nc->drain();
+```
+
+`WatchConfig` changes what is delivered:
+- `includeHistory` delivers every stored revision of the watched keys, not only the latest ones;
+- `updatesOnly` delivers only the changes made after the watch has started, `awaitInitialValues()` returns at once;
+- `metaOnly` delivers entries without values, `Entry::$size` still has the size of the value;
+- `ignoreDeletes` does not pass delete and purge markers to the handler;
+- `resumeFromRevision` delivers every revision starting from the given one, e.g. to continue a watch from the last handled revision.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+use Thesis\Nats\JetStream\KeyValue\Entry;
+use Thesis\Nats\JetStream\KeyValue\WatchConfig;
+use function Amp\delay;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
+$kv = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'configs',
+    history: 10,
+));
+
+$kv->put('app.env', 'prod');
+$kv->put('app.env', 'dev');
+
+$watcher = $kv->watch(
+    static function (Entry $entry): void {
+        dump("{$entry->key}@{$entry->revision}: {$entry->value}");
+    },
+    keys: 'app.*',
+    config: new WatchConfig(includeHistory: true, ignoreDeletes: true),
+);
+
+$watcher->awaitInitialValues();
+$watcher->stop();
+
+$watcher = $kv->watch(
+    static function (Entry $entry): void {
+        dump("{$entry->key} changed to {$entry->value}");
+    },
+    config: new WatchConfig(updatesOnly: true),
+);
+
+$kv->put('app.env', 'test');
+
+delay(1);
+$watcher->stop();
+
+$nc->drain();
+```
+
+#### Mirrors and sources
+
+A bucket with a `mirror` is a read-only copy of another bucket: it gets every entry of the origin, and a `put()` to the mirror goes to the origin.
+A bucket with `sources` gets the entries of the source buckets next to its own ones. The sources and the mirror are named by the bucket name.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Thesis\Nats;
+use Thesis\Nats\JetStream\Api\StreamSource;
+use Thesis\Nats\JetStream\KeyValue\BucketConfig;
+use function Amp\delay;
+
+$nc = new Nats\Client(Nats\Config::default());
+$js = $nc->jetStream();
+
+$eu = $js->createOrUpdateKeyValue(new BucketConfig(bucket: 'configs-eu'));
+$us = $js->createOrUpdateKeyValue(new BucketConfig(bucket: 'configs-us'));
+
+$replica = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'configs-eu-replica',
+    mirror: new StreamSource('configs-eu'),
+));
+
+$all = $js->createOrUpdateKeyValue(new BucketConfig(
+    bucket: 'configs-all',
+    sources: [
+        new StreamSource('configs-eu'),
+        new StreamSource('configs-us'),
+    ],
+));
+
+$eu->put('eu.dsn', 'mysql:host=10.0.0.1');
+$us->put('us.dsn', 'mysql:host=10.1.0.1');
+
+delay(1); // mirrors and sources are copied asynchronously
+
+dump(
+    $replica->get('eu.dsn')?->value,
+    $all->get('eu.dsn')?->value,
+    $all->get('us.dsn')?->value,
+);
 
 $nc->drain();
 ```
diff --git a/examples/keyvalue/bucket.php b/examples/keyvalue/bucket.php
index 8ec8028..ff0743b 100644
--- a/examples/keyvalue/bucket.php
+++ b/examples/keyvalue/bucket.php
@@ -13,7 +13,7 @@
 
 $kv = $js->createOrUpdateKeyValue(new KeyValue\BucketConfig('profiles'));
 
-$subscription = $kv->watch(static function (KeyValue\Entry $entry): void {
+$watcher = $kv->watch(static function (KeyValue\Entry $entry): void {
     if ($entry->state === KeyValue\EntryState::Created) {
         dump("Entry '{$entry->key}' in kv '{$entry->bucket}' was inserted.");
     } else {
@@ -28,5 +28,5 @@
 $kv->delete('user.1');
 
 trapSignal([\SIGINT, \SIGTERM]);
-$subscription->stop();
-$subscription->awaitCompletion();
+$watcher->stop();
+$watcher->awaitCompletion();
diff --git a/src/Header/ConsumerStalled.php b/src/Header/ConsumerStalled.php
new file mode 100644
index 0000000..da8d1c3
--- /dev/null
+++ b/src/Header/ConsumerStalled.php
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Header;
+
+/**
+ * Sent with an idle heartbeat of a push consumer that waits for the reply to its flow control request,
+ * the value is the subject to reply to.
+ *
+ * @api
+ */
+final readonly class ConsumerStalled
+{
+    private const string HEADER = 'Nats-Consumer-Stalled';
+
+    public static function header(): ScalarKey
+    {
+        return ScalarKey::string(self::HEADER);
+    }
+}
diff --git a/src/JetStream.php b/src/JetStream.php
index b935f5c..3a5b45c 100644
--- a/src/JetStream.php
+++ b/src/JetStream.php
@@ -8,6 +8,7 @@
 use Thesis\Nats\Exception\FeatureIsNotSupported;
 use Thesis\Nats\Exception\NoServerResponse;
 use Thesis\Nats\Exception\RequestHasNoResponders;
+use Thesis\Nats\Exception\StreamNameAlreadyInUse;
 use Thesis\Nats\Exception\StreamNotFound;
 use Thesis\Nats\Internal\Id;
 use Thesis\Nats\Internal\Validation\Validate;
@@ -21,6 +22,7 @@
 use Thesis\Nats\Json\NativeEncoder;
 use Thesis\Nats\Serialization\Serializer;
 use Thesis\Nats\Serialization\ValinorSerializer;
+use Thesis\Time\TimeSpan;
 use function Amp\delay;
 
 /**
@@ -28,6 +30,9 @@
  */
 final readonly class JetStream
 {
+    private const string KV_STREAM_PREFIX = 'KV_';
+    private const int KV_MAX_DUPLICATE_WINDOW_MINUTES = 2;
+
     private Api\Router $router;
 
     /**
@@ -201,13 +206,17 @@ public function purgeStream(
         ?int $sequence = null,
         ?int $keep = null,
         ?string $subject = null,
+        ?Cancellation $cancellation = null,
     ): Api\StreamPurged {
-        return $this->request(new Api\PurgeStreamRequest(
-            name: $name,
-            sequence: $sequence,
-            keep: $keep,
-            subject: $subject,
-        ));
+        return $this->request(
+            new Api\PurgeStreamRequest(
+                name: $name,
+                sequence: $sequence,
+                keep: $keep,
+                subject: $subject,
+            ),
+            $cancellation,
+        );
     }
 
     /**
@@ -289,8 +298,9 @@ public function updateConsumer(
     public function createOrUpdateConsumer(
         string $stream,
         Api\ConsumerConfig $config = new Api\ConsumerConfig(),
+        ?Cancellation $cancellation = null,
     ): JetStream\Consumer {
-        return $this->upsertConsumer($stream, $config);
+        return $this->upsertConsumer($stream, $config, cancellation: $cancellation);
     }
 
     /**
@@ -313,12 +323,18 @@ public function consumerInfo(
      * @param non-empty-string $consumer
      * @throws NatsException
      */
-    public function deleteConsumer(string $stream, string $consumer): Api\ConsumerDeleted
-    {
-        return $this->request(new Api\DeleteConsumerRequest(
-            stream: $stream,
-            consumer: $consumer,
-        ));
+    public function deleteConsumer(
+        string $stream,
+        string $consumer,
+        ?Cancellation $cancellation = null,
+    ): Api\ConsumerDeleted {
+        return $this->request(
+            new Api\DeleteConsumerRequest(
+                stream: $stream,
+                consumer: $consumer,
+            ),
+            $cancellation,
+        );
     }
 
     /**
@@ -398,57 +414,36 @@ public function consumerList(string $stream): iterable
     }
 
     /**
+     * Creates the bucket, fails with {@see StreamNameAlreadyInUse} if it exists with another config.
+     *
      * @throws NatsException
      */
-    public function createOrUpdateKeyValue(KeyValue\BucketConfig $config): KeyValue\Bucket
+    public function createKeyValue(KeyValue\BucketConfig $config): KeyValue\Bucket
     {
-        $allowMessageTtl = false;
-        $subjectDeleteMarkerTtl = null;
-
-        if ($config->limitMarkerTtl !== null) {
-            $info = $this->accountInfo();
-
-            if ($info->api->level < 1) {
-                throw FeatureIsNotSupported::forLimitMarkerTtl();
-            }
-
-            $allowMessageTtl = true;
-            $subjectDeleteMarkerTtl = $config->limitMarkerTtl;
-        }
-
-        $stream = $this->createOrUpdateStream(new Api\StreamConfig(
-            name: "KV_{$config->bucket}",
-            description: $config->description,
-            subjects: ["\$KV.{$config->bucket}.>"],
-            discard: Api\DiscardPolicy::New,
-            maxConsumers: -1,
-            maxMessages: -1,
-            maxBytes: $config->maxBytes,
-            maxAge: $config->ttl,
-            maxMessagesPerSubject: $config->history,
-            maxMessageSize: $config->maxValueSize,
-            storageType: $config->storageType,
-            replicas: $config->replicas,
-            duplicateWindow: $config->ttl,
-            placement: $config->placement,
-            denyDelete: true,
-            allowRollup: true,
-            compression: $config->compression ? Api\StoreCompression::S2 : Api\StoreCompression::None,
-            rePublish: $config->rePublish,
-            allowDirect: true,
-            allowMessageTtl: $allowMessageTtl,
-            subjectDeleteMarkerTtl: $subjectDeleteMarkerTtl,
-        ));
+        return $this->setupKeyValue($config->bucket, $this->createStream($this->keyValueStreamConfig($config)));
+    }
 
-        return new KeyValue\Bucket(
-            name: $config->bucket,
-            js: $this,
-            stream: $stream,
-            prefix: "\$KV.{$config->bucket}.",
-            jsPrefix: $this->router->prefix() !== Api\Router::DEFAULT_PREFIX ? $this->router->prefix() : null,
+    /**
+     * Updates the config of the bucket, fails with {@see StreamNotFound} if there is no bucket.
+     *
+     * @throws NatsException
+     */
+    public function updateKeyValue(KeyValue\BucketConfig $config): KeyValue\Bucket
+    {
+        return $this->setupKeyValue(
+            $config->bucket,
+            $this->setupStream($this->updateStream($this->keyValueStreamConfig($config))),
         );
     }
 
+    /**
+     * @throws NatsException
+     */
+    public function createOrUpdateKeyValue(KeyValue\BucketConfig $config): KeyValue\Bucket
+    {
+        return $this->setupKeyValue($config->bucket, $this->createOrUpdateStream($this->keyValueStreamConfig($config)));
+    }
+
     /**
      * @param non-empty-string $bucket
      * @throws NatsException
@@ -458,7 +453,7 @@ public function deleteKeyValue(string $bucket): void
         Validate::bucket($bucket);
 
         try {
-            $this->deleteStream("KV_{$bucket}");
+            $this->deleteStream(self::kvStreamName($bucket));
         } catch (StreamNotFound) {
         }
     }
@@ -470,19 +465,9 @@ public function keyValue(string $bucket): ?KeyValue\Bucket
     {
         Validate::bucket($bucket);
 
-        $stream = $this->stream("KV_{$bucket}");
+        $stream = $this->stream(self::kvStreamName($bucket));
 
-        if ($stream !== null) {
-            return new KeyValue\Bucket(
-                name: $bucket,
-                js: $this,
-                stream: $stream,
-                prefix: "\$KV.{$bucket}.",
-                jsPrefix: $this->router->prefix() !== Api\Router::DEFAULT_PREFIX ? $this->router->prefix() : null,
-            );
-        }
-
-        return null;
+        return $stream !== null ? $this->setupKeyValue($bucket, $stream) : null;
     }
 
     /**
@@ -684,7 +669,7 @@ public function rawPublish(string $subject, mixed $payload = null, ?string $repl
      * @internal
      * @param non-empty-string $endpoint
      */
-    public function rawRequest(string $endpoint, mixed $payload = null): Message
+    public function rawRequest(string $endpoint, mixed $payload = null, ?Cancellation $cancellation = null): Message
     {
         return $this->nats
             ->rawRequest(
@@ -692,6 +677,7 @@ public function rawRequest(string $endpoint, mixed $payload = null): Message
                 message: new Message(
                     payload: $payload !== null ? $this->encoder->encode($payload) : null,
                 ),
+                cancellation: $cancellation,
             )
             ->message;
     }
@@ -703,33 +689,189 @@ public function rawRequest(string $endpoint, mixed $payload = null): Message
      * @return T
      * @throws NatsException
      */
-    public function request(Api\Request $request): mixed
+    public function request(Api\Request $request, ?Cancellation $cancellation = null): mixed
     {
         return $this->mapResponse(
-            message: $this->rawRequest($request->endpoint(), $request->payload()),
+            message: $this->rawRequest($request->endpoint(), $request->payload(), $cancellation),
             type: $request->type(),
         );
     }
 
+    /**
+     * @throws NatsException
+     */
+    private function keyValueStreamConfig(KeyValue\BucketConfig $config): Api\StreamConfig
+    {
+        $allowMessageTtl = false;
+        $subjectDeleteMarkerTtl = null;
+
+        if ($config->limitMarkerTtl !== null) {
+            $info = $this->accountInfo();
+
+            if ($info->api->level < 1) {
+                throw FeatureIsNotSupported::forLimitMarkerTtl();
+            }
+
+            $allowMessageTtl = true;
+            $subjectDeleteMarkerTtl = $config->limitMarkerTtl;
+        }
+
+        // As in nats.go: the server default of two minutes, unless the entries expire earlier.
+        $duplicateWindow = TimeSpan::fromMinutes(self::KV_MAX_DUPLICATE_WINDOW_MINUTES);
+        if ($config->ttl !== null && $config->ttl->isPositive() && $config->ttl->isLessThan($duplicateWindow)) {
+            $duplicateWindow = $config->ttl;
+        }
+
+        $subjects = ["\$KV.{$config->bucket}.>"];
+        $mirror = null;
+        $sources = null;
+
+        if ($config->mirror !== null) {
+            // A mirror keeps the subjects of the origin bucket and takes no subjects of its own.
+            $mirror = self::kvStreamSource(
+                source: $config->mirror,
+                name: self::kvStreamName(self::kvBucketName($config->mirror->name)),
+                subjectTransforms: $config->mirror->subjectTransforms,
+            );
+            $subjects = null;
+        } elseif ($config->sources !== null && $config->sources !== []) {
+            $sources = [];
+
+            foreach ($config->sources as $source) {
+                $sourceBucket = self::kvBucketName($source->name);
+
+                // The entries of the source bucket become the entries of this one, unless it is the same bucket in another domain.
+                $subjectTransforms = $source->external === null || $sourceBucket !== $config->bucket
+                    ? [new Api\SubjectTransformConfig("\$KV.{$sourceBucket}.>", "\$KV.{$config->bucket}.>")]
+                    : $source->subjectTransforms;
+
+                $sources[] = self::kvStreamSource(
+                    source: $source,
+                    name: self::kvStreamName($sourceBucket),
+                    subjectTransforms: $subjectTransforms,
+                );
+            }
+        }
+
+        return new Api\StreamConfig(
+            name: self::kvStreamName($config->bucket),
+            description: $config->description,
+            subjects: $subjects,
+            discard: Api\DiscardPolicy::New,
+            maxConsumers: -1,
+            maxMessages: -1,
+            maxBytes: $config->maxBytes,
+            maxAge: $config->ttl,
+            maxMessagesPerSubject: $config->history,
+            maxMessageSize: $config->maxValueSize,
+            storageType: $config->storageType,
+            replicas: $config->replicas,
+            duplicateWindow: $duplicateWindow,
+            placement: $config->placement,
+            mirror: $mirror,
+            sources: $sources,
+            denyDelete: true,
+            allowRollup: true,
+            compression: $config->compression ? Api\StoreCompression::S2 : Api\StoreCompression::None,
+            rePublish: $config->rePublish,
+            allowDirect: true,
+            mirrorDirect: $mirror !== null,
+            allowMessageTtl: $allowMessageTtl,
+            subjectDeleteMarkerTtl: $subjectDeleteMarkerTtl,
+        );
+    }
+
+    /**
+     * @param non-empty-string $bucket
+     */
+    private function setupKeyValue(string $bucket, JetStream\Stream $stream): KeyValue\Bucket
+    {
+        $jsPrefix = $this->router->prefix() !== Api\Router::DEFAULT_PREFIX ? $this->router->prefix() : '';
+
+        $prefix = "\$KV.{$bucket}.";
+        $putPrefix = "{$jsPrefix}{$prefix}";
+
+        $mirror = $stream->info->config->mirror;
+
+        if ($mirror !== null) {
+            // The entries of a mirror are stored under the subjects of the origin bucket, and are put to the origin.
+            $origin = self::kvBucketName($mirror->name);
+            $prefix = "\$KV.{$origin}.";
+            $api = $mirror->external?->api;
+            $putPrefix = $api !== null ? "{$api}.{$prefix}" : "{$jsPrefix}{$prefix}";
+        }
+
+        return new KeyValue\Bucket(
+            name: $bucket,
+            js: $this,
+            stream: $stream,
+            prefix: $prefix,
+            putPrefix: $putPrefix,
+        );
+    }
+
+    /**
+     * @param non-empty-string $name a bucket name or the name of its stream
+     * @return non-empty-string
+     */
+    private static function kvBucketName(string $name): string
+    {
+        $bucket = str_starts_with($name, self::KV_STREAM_PREFIX) ? substr($name, \strlen(self::KV_STREAM_PREFIX)) : $name;
+
+        return $bucket !== '' ? $bucket : $name;
+    }
+
+    /**
+     * @param non-empty-string $bucket
+     * @return non-empty-string
+     */
+    private static function kvStreamName(string $bucket): string
+    {
+        return self::KV_STREAM_PREFIX . $bucket;
+    }
+
+    /**
+     * @param non-empty-string $name
+     * @param ?list<Api\SubjectTransformConfig> $subjectTransforms
+     */
+    private static function kvStreamSource(Api\StreamSource $source, string $name, ?array $subjectTransforms): Api\StreamSource
+    {
+        return new Api\StreamSource(
+            name: $name,
+            startSeq: $source->startSeq,
+            startTime: $source->startTime,
+            filterSubject: $source->filterSubject,
+            subjectTransforms: $subjectTransforms,
+            external: $source->external,
+        );
+    }
+
     /**
      * @param non-empty-string $stream
      * @param ?Api\CreateConsumerRequest::ACTION_* $action
      * @throws NatsException
      */
-    private function upsertConsumer(string $stream, Api\ConsumerConfig $config, ?string $action = null): JetStream\Consumer
-    {
+    private function upsertConsumer(
+        string $stream,
+        Api\ConsumerConfig $config,
+        ?string $action = null,
+        ?Cancellation $cancellation = null,
+    ): JetStream\Consumer {
         $consumerName = $config->name ?? $config->durableName;
 
         if ($consumerName === null || $consumerName === '') {
             $consumerName = Id\generateUniqueId(10);
         }
 
-        $info = $this->request(new Api\CreateConsumerRequest(
-            stream: $stream,
-            consumer: $consumerName,
-            config: $config,
-            action: $action,
-        ));
+        $info = $this->request(
+            new Api\CreateConsumerRequest(
+                stream: $stream,
+                consumer: $consumerName,
+                config: $config,
+                action: $action,
+            ),
+            $cancellation,
+        );
 
         return $this->setupConsumer($info);
     }
diff --git a/src/JetStream/Api/Mapping/FixStructureSource.php b/src/JetStream/Api/Mapping/FixStructureSource.php
index 864125c..da81be7 100644
--- a/src/JetStream/Api/Mapping/FixStructureSource.php
+++ b/src/JetStream/Api/Mapping/FixStructureSource.php
@@ -4,12 +4,30 @@
 
 namespace Thesis\Nats\JetStream\Api\Mapping;
 
+use Thesis\Nats\JetStream\Api\StreamConfig;
+
 /**
  * @internal
  * @template-implements \IteratorAggregate<non-empty-string, mixed>
  */
 final readonly class FixStructureSource implements \IteratorAggregate
 {
+    /**
+     * The keys of a stream config the server sends that differ from the property names of {@see StreamConfig}.
+     */
+    private const array STREAM_CONFIG_KEYS = [
+        'max_msgs' => 'max_messages',
+        'max_msgs_per_subject' => 'max_messages_per_subject',
+        'max_msg_size' => 'max_message_size',
+        'storage' => 'storage_type',
+        'num_replicas' => 'replicas',
+        'allow_rollup_hdrs' => 'allow_rollup',
+        'allow_msg_ttl' => 'allow_message_ttl',
+        'allow_msg_counter' => 'allow_message_counter',
+        'allow_atomic' => 'allow_atomic_publish',
+        'republish' => 're_publish',
+    ];
+
     /** @var array<non-empty-string, mixed> */
     private array $data;
 
@@ -18,6 +36,9 @@
      */
     public function __construct(array $data)
     {
+        /** @var array<non-empty-string, mixed> $data */
+        $data = self::fixStreamConfigs($data);
+
         if (!isset($data['error'])) {
             $data['response'] = $data;
         }
@@ -29,4 +50,43 @@ public function getIterator(): \Traversable
     {
         yield from $this->data;
     }
+
+    /**
+     * @param array<mixed> $data
+     * @return array<mixed>
+     */
+    private static function fixStreamConfigs(array $data): array
+    {
+        foreach ($data as $key => $value) {
+            if (!\is_array($value)) {
+                continue;
+            }
+
+            $value = self::fixStreamConfigs($value);
+
+            // A stream config always has the retention policy, a consumer config never does.
+            if ($key === 'config' && isset($value['retention'])) {
+                $value = self::renameStreamConfigKeys($value);
+            }
+
+            $data[$key] = $value;
+        }
+
+        return $data;
+    }
+
+    /**
+     * @param array<mixed> $config
+     * @return array<mixed>
+     */
+    private static function renameStreamConfigKeys(array $config): array
+    {
+        $renamed = [];
+
+        foreach ($config as $key => $value) {
+            $renamed[\is_string($key) ? self::STREAM_CONFIG_KEYS[$key] ?? $key : $key] = $value;
+        }
+
+        return $renamed;
+    }
 }
diff --git a/src/JetStream/Consumer.php b/src/JetStream/Consumer.php
index 185c1e7..b9aba95 100644
--- a/src/JetStream/Consumer.php
+++ b/src/JetStream/Consumer.php
@@ -96,11 +96,12 @@ public function actualInfo(): Api\ConsumerInfo
     /**
      * @throws NatsException
      */
-    public function delete(): Api\ConsumerDeleted
+    public function delete(?Cancellation $cancellation = null): Api\ConsumerDeleted
     {
         return $this->js->deleteConsumer(
             stream: $this->stream,
             consumer: $this->name,
+            cancellation: $cancellation,
         );
     }
 
diff --git a/src/JetStream/Internal/IdleListener.php b/src/JetStream/Internal/IdleListener.php
new file mode 100644
index 0000000..d95aea4
--- /dev/null
+++ b/src/JetStream/Internal/IdleListener.php
@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\JetStream\Internal;
+
+/**
+ * A push consumer handler that wants to know when the server has nothing more to deliver.
+ *
+ * @internal
+ */
+interface IdleListener
+{
+    /**
+     * Called on an idle heartbeat: the server sends it only when it has not delivered anything for the heartbeat interval
+     * and does not wait for flow control, so every message before it has been handled.
+     */
+    public function idle(): void;
+}
diff --git a/src/JetStream/Internal/PushMessageHandler.php b/src/JetStream/Internal/PushMessageHandler.php
index 7921827..ae86235 100644
--- a/src/JetStream/Internal/PushMessageHandler.php
+++ b/src/JetStream/Internal/PushMessageHandler.php
@@ -7,6 +7,7 @@
 use Thesis\Nats\Client;
 use Thesis\Nats\Delivery as NatsDelivery;
 use Thesis\Nats\Description;
+use Thesis\Nats\Header;
 use Thesis\Nats\JetStream\Delivery as JetStreamDelivery;
 use Thesis\Nats\Message;
 use Thesis\Nats\Status;
@@ -56,6 +57,13 @@ private function doHandleDelivery(
                 case [Status::Control, Description::FlowControl]: // @phpstan-ignore switch.type
                     $delivery->reply(new Message());
                     break;
+                case [Status::Control, Description::IdleHeartbeat]: // @phpstan-ignore switch.type
+                    $stalled = $delivery->message->headers?->exists(Header\ConsumerStalled::header()) ?? false;
+
+                    if (!$stalled && $this->handler instanceof IdleListener) {
+                        $this->handler->idle();
+                    }
+                    break;
                 case [Status::Conflict, Description::ConsumerDeleted]: // @phpstan-ignore switch.type
                     $subscription->stop();
                     break;
diff --git a/src/JetStream/KeyValue/Bucket.php b/src/JetStream/KeyValue/Bucket.php
index 2cb0a9d..ae4b3cc 100644
--- a/src/JetStream/KeyValue/Bucket.php
+++ b/src/JetStream/KeyValue/Bucket.php
@@ -5,18 +5,23 @@
 namespace Thesis\Nats\JetStream\KeyValue;
 
 use Amp\Cancellation;
+use Amp\CancelledException;
+use Thesis\Nats\Exception\InvalidKey;
+use Thesis\Nats\Exception\WrongLastSequence;
 use Thesis\Nats\Header;
 use Thesis\Nats\Headers;
 use Thesis\Nats\Internal\Id;
 use Thesis\Nats\Internal\Validation\Validate;
 use Thesis\Nats\JetStream;
+use Thesis\Nats\JetStream\Api\AckPolicy;
 use Thesis\Nats\JetStream\Api\DeliverPolicy;
 use Thesis\Nats\JetStream\Api\ReplayPolicy;
-use Thesis\Nats\JetStream\Delivery;
 use Thesis\Nats\Message;
 use Thesis\Nats\NatsException;
+use Thesis\Nats\PublishOptions;
 use Thesis\Nats\Subscription;
 use Thesis\Time\TimeSpan;
+use function Amp\async;
 
 /**
  * @api
@@ -25,22 +30,33 @@
 {
     private const string ALL_KEYS = '>';
 
+    /** How often the server sends heartbeats to an idle watch, as nats.go does for ordered consumers. */
+    private const int WATCH_HEARTBEAT_SECONDS = 5;
+
+    /** The age of the delete and purge markers {@see purgeDeletes()} removes by default, as in nats.go. */
+    private const int PURGE_DELETES_OLDER_THAN_MINUTES = 30;
+
     /**
+     * @internal
      * @param non-empty-string $name
-     * @param non-empty-string $prefix
-     * @param ?non-empty-string $jsPrefix
+     * @param non-empty-string $prefix the subject prefix of the stored entries: of the origin bucket for a mirror
+     * @param non-empty-string $putPrefix the subject prefix to publish entries to, with the JetStream API prefix of a domain if any
      */
     public function __construct(
         public string $name,
         private JetStream $js,
         private JetStream\Stream $stream,
         private string $prefix,
-        private ?string $jsPrefix = null,
+        private string $putPrefix,
     ) {}
 
     /**
+     * Returns the entry of the latest revision of the key or of the given one, null if there is none.
+     * A deleted or purged key has an entry too, with {@see Entry::$state} telling so and no value.
+     *
      * @param non-empty-string $key
      * @param non-negative-int $revision
+     * @throws InvalidKey
      */
     public function get(string $key, int $revision = 0): ?Entry
     {
@@ -57,54 +73,81 @@ public function get(string $key, int $revision = 0): ?Entry
             return null;
         }
 
-        $state = EntryState::fromKVOperation($message->headers->get(Header\KvOperation::header()));
-
         return new Entry(
-            bucket: $this->stream->name,
+            bucket: $this->name,
             key: $key,
             created: $message->headers->get(Header\Timestamp::Header) ?? new \DateTimeImmutable(),
             revision: $message->headers->get(Header\Sequence::header()) ?? 1,
             size: \strlen($message->payload ?? ''),
             value: $message->payload,
-            state: $state,
+            state: EntryState::fromHeaders($message->headers),
         );
     }
 
     /**
      * @param non-empty-string $key
-     * @return non-negative-int
+     * @param ?TimeSpan $ttl how long to keep the entry, requires {@see BucketConfig::$limitMarkerTtl}
+     * @return non-negative-int the revision of the entry
+     * @throws NatsException
      */
-    public function put(string $key, ?string $value = null, ?Headers $headers = null): int
-    {
-        return $this->js
-            ->publish($this->prefixedSubject($key), new Message($value, $headers))
-            ->seq ?? throw new \LogicException('Sequence expected on kv publish');
+    public function put(
+        string $key,
+        ?string $value = null,
+        ?Headers $headers = null,
+        ?TimeSpan $ttl = null,
+    ): int {
+        return $this->publish($key, new Message($value, $headers), new PublishOptions(ttl: $ttl));
     }
 
     /**
+     * Puts the value only if the key has no value: it was never put, or it was deleted or purged.
+     * Otherwise, the server rejects the value with {@see WrongLastSequence}.
+     *
      * @param non-empty-string $key
-     * @param non-negative-int $revision
-     * @return non-negative-int
+     * @param ?TimeSpan $ttl how long to keep the entry, requires {@see BucketConfig::$limitMarkerTtl}
+     * @return non-negative-int the revision of the entry
+     * @throws WrongLastSequence if the key has a value
+     * @throws NatsException
      */
-    public function update(
+    public function create(
         string $key,
-        int $revision,
         ?string $value = null,
         ?TimeSpan $ttl = null,
     ): int {
-        $headers = (new Headers())
-            ->with(Header\ExpectedLastSubjSeq::Header, $revision);
+        try {
+            return $this->update($key, 0, $value, $ttl);
+        } catch (WrongLastSequence $e) {
+            // The last message of the key may be a delete or purge marker: the key has no value then.
+            $entry = $this->get($key);
+
+            if ($entry !== null && $entry->state->isMarker()) {
+                return $this->update($key, $entry->revision, $value, $ttl);
+            }
 
-        if ($ttl !== null) {
-            $headers = $headers->with(Header\MsgTtl::Header, $ttl);
+            throw $e;
         }
+    }
 
-        return $this->js
-            ->publish($this->prefixedSubject($key), new Message(
-                payload: $value,
-                headers: $headers,
-            ))
-            ->seq ?? throw new \LogicException('Sequence expected on kv update');
+    /**
+     * Puts the value only if the latest revision of the key is the given one.
+     *
+     * @param non-empty-string $key
+     * @param non-negative-int $revision
+     * @param ?TimeSpan $ttl how long to keep the entry, requires {@see BucketConfig::$limitMarkerTtl}
+     * @return non-negative-int the revision of the entry
+     * @throws WrongLastSequence if the latest revision differs
+     * @throws NatsException
+     */
+    public function update(
+        string $key,
+        int $revision,
+        ?string $value = null,
+        ?TimeSpan $ttl = null,
+    ): int {
+        return $this->publish($key, new Message($value), new PublishOptions(
+            expectedLastSubjectSequence: $revision,
+            ttl: $ttl,
+        ));
     }
 
     /**
@@ -114,54 +157,158 @@ public function update(
      */
     public function delete(string $key, ?int $revision = null): void
     {
-        $headers = (new Headers())
-            ->with(Header\KvOperation::header(), Header\KvOperation::OP_DEL);
-
-        if ($revision !== null) {
-            $headers = $headers->with(Header\ExpectedLastSubjSeq::Header, $revision);
-        }
-
-        $this->js->publish($this->prefixedSubject($key), new Message(
-            headers: $headers,
-        ));
+        $this->publish(
+            $key,
+            new Message(
+                headers: (new Headers())
+                    ->with(Header\KvOperation::header(), Header\KvOperation::OP_DEL),
+            ),
+            new PublishOptions(expectedLastSubjectSequence: $revision),
+        );
     }
 
     /**
      * @param non-empty-string $key
      * @param ?non-negative-int $revision
+     * @throws NatsException
      */
     public function purge(
         string $key,
         ?int $revision = null,
         ?TimeSpan $ttl = null,
     ): void {
-        $headers = (new Headers())
-            ->with(Header\KvOperation::header(), Header\KvOperation::OP_PURGE)
-            ->with(Header\MsgRollup::header(), Header\MsgRollup::ROLLUP_SUBJECT);
+        $this->publish(
+            $key,
+            new Message(
+                headers: (new Headers())
+                    ->with(Header\KvOperation::header(), Header\KvOperation::OP_PURGE)
+                    ->with(Header\MsgRollup::header(), Header\MsgRollup::ROLLUP_SUBJECT),
+            ),
+            new PublishOptions(
+                expectedLastSubjectSequence: $revision,
+                ttl: $ttl,
+            ),
+        );
+    }
+
+    /**
+     * Calls the handler with the latest entry of every watched key and then with every change.
+     * {@see Watcher::awaitInitialValues()} tells when the entries the bucket had at the start have been handled,
+     * {@see WatchConfig} changes what is delivered.
+     *
+     * @param callable(Entry, Subscription): void $handler
+     * @param list<non-empty-string>|non-empty-string $keys keys or patterns with wildcards, all keys if empty
+     * @throws NatsException
+     */
+    public function watch(
+        callable $handler,
+        string|array $keys = [],
+        WatchConfig $config = new WatchConfig(),
+        ?Cancellation $cancellation = null,
+    ): Watcher {
+        return $this->startWatch($handler, $keys, $config, cancellation: $cancellation);
+    }
 
-        if ($revision !== null) {
-            $headers = $headers->with(Header\ExpectedLastSubjSeq::Header, $revision);
+    /**
+     * Returns the keys that have a value, i.e. are not deleted or purged.
+     *
+     * @param list<non-empty-string>|non-empty-string $filters keys or patterns with wildcards, all keys if empty
+     * @return list<non-empty-string>
+     * @throws NatsException
+     */
+    public function keys(string|array $filters = [], ?Cancellation $cancellation = null): array
+    {
+        return array_map(
+            static fn(Entry $entry): string => $entry->key,
+            $this->initialEntries($filters, new WatchConfig(ignoreDeletes: true, metaOnly: true), $cancellation),
+        );
+    }
+
+    /**
+     * Returns every stored revision of the key, from the oldest, delete and purge markers included.
+     *
+     * @param non-empty-string $key a key or a pattern with wildcards
+     * @return list<Entry>
+     * @throws NatsException
+     */
+    public function history(string $key, ?Cancellation $cancellation = null): array
+    {
+        return $this->initialEntries($key, new WatchConfig(includeHistory: true), $cancellation);
+    }
+
+    /**
+     * Removes the revisions of the deleted and purged keys. The markers older than $olderThan are removed as well,
+     * the recent ones are kept, so that the watches that are behind still see the keys are gone.
+     *
+     * @param ?TimeSpan $olderThan 30 minutes by default, zero or negative to remove all markers
+     * @throws NatsException
+     */
+    public function purgeDeletes(?TimeSpan $olderThan = null, ?Cancellation $cancellation = null): void
+    {
+        $olderThan ??= TimeSpan::fromMinutes(self::PURGE_DELETES_OLDER_THAN_MINUTES);
+
+        $markers = array_filter(
+            $this->initialEntries([], new WatchConfig(metaOnly: true), $cancellation),
+            static fn(Entry $entry): bool => $entry->state->isMarker(),
+        );
+
+        $now = new \DateTimeImmutable();
+
+        foreach ($markers as $marker) {
+            $recent = $olderThan->isPositive() && TimeSpan::diff($now, $marker->created)->isLessThan($olderThan);
+
+            $this->stream->purge(
+                keep: $recent ? 1 : null,
+                subject: "{$this->prefix}{$marker->key}",
+                cancellation: $cancellation,
+            );
         }
+    }
+
+    /**
+     * @throws NatsException
+     */
+    public function status(): BucketStatus
+    {
+        return new BucketStatus($this->name, $this->stream->actualInfo());
+    }
 
-        if ($ttl !== null) {
-            $headers = $headers->with(Header\MsgTtl::Header, $ttl);
+    /**
+     * @param list<non-empty-string>|non-empty-string $filters
+     * @return list<Entry>
+     * @throws NatsException
+     */
+    private function initialEntries(string|array $filters, WatchConfig $config, ?Cancellation $cancellation): array
+    {
+        $entries = new Internal\EntryCollector();
+
+        $watcher = $this->startWatch($entries, $filters, $config, initialValuesOnly: true, cancellation: $cancellation);
+
+        try {
+            $watcher->awaitInitialValues($cancellation);
+        } finally {
+            try {
+                $watcher->stop($cancellation);
+            } catch (CancelledException) {
+                // The server removes the ephemeral consumer itself once the subscription is gone.
+            }
         }
 
-        $this->js->publish($this->prefixedSubject($key), new Message(
-            headers: $headers,
-        ));
+        return $entries->entries();
     }
 
     /**
      * @param callable(Entry, Subscription): void $handler
      * @param list<non-empty-string>|non-empty-string $keys
+     * @throws NatsException
      */
-    public function watch(
+    private function startWatch(
         callable $handler,
-        string|array $keys = [],
-        WatchConfig $config = new WatchConfig(),
+        string|array $keys,
+        WatchConfig $config,
+        bool $initialValuesOnly = false,
         ?Cancellation $cancellation = null,
-    ): Subscription {
+    ): Watcher {
         if (!\is_array($keys)) {
             $keys = [$keys];
         }
@@ -175,78 +322,72 @@ public function watch(
             $keys !== [] ? $keys : [self::ALL_KEYS],
         );
 
-        $name = $this->name;
-        $prefix = $this->prefix;
-
-        return $this->stream
-            ->createOrUpdateConsumer(new JetStream\Api\ConsumerConfig(
-                description: 'kv watch consumer',
-                deliverPolicy: DeliverPolicy::New,
-                deliverSubject: Id\generateInboxId(),
-                replayPolicy: ReplayPolicy::Instant,
-                headersOnly: $config->headersOnly,
-                filterSubjects: $keys,
-            ))
-            ->push(
-                static function (Delivery $delivery, Subscription $subscription) use (
-                    $config,
-                    $handler,
-                    $name,
-                    $prefix,
-                ): void {
-                    $metadata = $delivery->metadata;
-                    if ($metadata === null) {
-                        return;
-                    }
-
-                    $key = substr($delivery->subject, \strlen($prefix));
-                    if ($key === '') {
-                        return;
-                    }
-
-                    $state = EntryState::fromKVOperation($delivery->message->headers?->get(Header\KvOperation::header()));
-
-                    if ($config->ignoreDeletes && \in_array($state, [EntryState::Deleted, EntryState::Purged], true)) {
-                        return;
-                    }
-
-                    $payload = $delivery->message->payload;
-
-                    $entry = new Entry(
-                        bucket: $name,
-                        key: $key,
-                        created: $metadata->timestamp,
-                        revision: max($metadata->streamSequence, 0),
-                        size: $payload !== null ? \strlen($payload) : (
-                            $delivery->message->headers?->get(Header\MsgSize::header()) ?? 0
-                        ),
-                        value: $payload,
-                        delta: $metadata->pending,
-                        state: $state,
-                    );
-
-                    $handler($entry, $subscription);
-                },
-                cancellation: $cancellation,
-            );
+        // The same precedence as in nats.go: the revision to resume from wins, then updates only.
+        [$deliverPolicy, $startSeq] = match (true) {
+            $config->resumeFromRevision !== null => [DeliverPolicy::ByStartSequence, $config->resumeFromRevision],
+            $config->updatesOnly => [DeliverPolicy::New, null],
+            $config->includeHistory => [DeliverPolicy::All, null],
+            default => [DeliverPolicy::LastPerSubject, null],
+        };
+
+        // An ordered consumer of nats.go: no acks, flow control with heartbeats, a replica in memory.
+        $consumerConfig = new JetStream\Api\ConsumerConfig(
+            description: 'kv watch consumer',
+            deliverPolicy: $deliverPolicy,
+            deliverSubject: Id\generateInboxId(),
+            startSeq: $startSeq,
+            ackPolicy: AckPolicy::None,
+            maxDeliver: 1,
+            flowControl: true,
+            idleHeartbeat: TimeSpan::fromSeconds(self::WATCH_HEARTBEAT_SECONDS),
+            replayPolicy: ReplayPolicy::Instant,
+            headersOnly: $config->metaOnly || $config->headersOnly,
+            replicas: 1,
+            memoryStorage: true,
+            filterSubjects: $keys,
+        );
+
+        $consumer = $this->stream->createOrUpdateConsumer($consumerConfig, $cancellation);
+
+        $initialValues = $config->updatesOnly
+            ? Internal\InitialValues::none()
+            : new Internal\InitialValues($consumer->info->numPending + $consumer->info->delivered->consumerSeq);
+
+        $subscription = $consumer->push(
+            new Internal\WatchHandler(
+                handler: $handler,
+                bucket: $this->name,
+                prefix: $this->prefix,
+                ignoreDeletes: $config->ignoreDeletes,
+                initialValues: $initialValues,
+                initialValuesOnly: $initialValuesOnly,
+            ),
+            cancellation: $cancellation,
+        );
+
+        // Awaited only together with the initial values: a failed watch must not be reported as unhandled.
+        $completion = async($subscription->awaitCompletion(...));
+        $completion->ignore();
+
+        return new Watcher(
+            subscription: $subscription,
+            consumer: $consumer,
+            initialValues: $initialValues->future(),
+            completion: $completion,
+        );
     }
 
     /**
-     * @param non-empty-string $cmd
-     * @return non-empty-string
+     * @param non-empty-string $key
+     * @return non-negative-int
+     * @throws NatsException
      */
-    private function prefixedSubject(string $cmd): string
+    private function publish(string $key, Message $message, PublishOptions $options): int
     {
-        Validate::key($cmd);
-
-        $subject = '';
-
-        if ($this->jsPrefix !== null) {
-            $subject .= $this->jsPrefix;
-        }
-
-        $subject .= "{$this->prefix}{$cmd}";
+        Validate::key($key);
 
-        return $subject;
+        return $this->js
+            ->publish("{$this->putPrefix}{$key}", $message, $options)
+            ->seq ?? throw new \LogicException('Sequence expected on kv publish');
     }
 }
diff --git a/src/JetStream/KeyValue/BucketStatus.php b/src/JetStream/KeyValue/BucketStatus.php
new file mode 100644
index 0000000..d43f420
--- /dev/null
+++ b/src/JetStream/KeyValue/BucketStatus.php
@@ -0,0 +1,71 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\JetStream\KeyValue;
+
+use Thesis\Nats\JetStream\Api\StoreCompression;
+use Thesis\Nats\JetStream\Api\StreamInfo;
+use Thesis\Time\TimeSpan;
+
+/**
+ * @api
+ */
+final readonly class BucketStatus
+{
+    public const string BACKING_STORE = 'JetStream';
+
+    /**
+     * The number of stored messages: every kept revision of every key, delete and purge markers included.
+     *
+     * @var non-negative-int
+     */
+    public int $values;
+
+    /** How many revisions of a key the bucket keeps. */
+    public int $history;
+
+    /** How long an entry is kept, null if forever. */
+    public ?TimeSpan $ttl;
+
+    /**
+     * The size of the stored messages.
+     *
+     * @var non-negative-int
+     */
+    public int $bytes;
+
+    /** @var non-empty-string */
+    public string $backingStore;
+
+    public bool $compressed;
+
+    /** How long the markers of expired keys are kept, null if the server does not place them. */
+    public ?TimeSpan $limitMarkerTtl;
+
+    /** @var array<string, string> */
+    public array $metadata;
+
+    /**
+     * @internal
+     * @param non-empty-string $bucket
+     */
+    public function __construct(
+        public string $bucket,
+        public StreamInfo $info,
+    ) {
+        $this->values = $info->state->messages;
+        $this->history = $info->config->maxMessagesPerSubject;
+        $this->ttl = self::nonZero($info->config->maxAge);
+        $this->bytes = $info->state->bytes;
+        $this->backingStore = self::BACKING_STORE;
+        $this->compressed = $info->config->compression !== StoreCompression::None;
+        $this->limitMarkerTtl = self::nonZero($info->config->subjectDeleteMarkerTtl);
+        $this->metadata = $info->config->metadata ?? [];
+    }
+
+    private static function nonZero(?TimeSpan $span): ?TimeSpan
+    {
+        return $span !== null && !$span->isZero() ? $span : null;
+    }
+}
diff --git a/src/JetStream/KeyValue/EntryState.php b/src/JetStream/KeyValue/EntryState.php
index 46b9c22..116c141 100644
--- a/src/JetStream/KeyValue/EntryState.php
+++ b/src/JetStream/KeyValue/EntryState.php
@@ -5,6 +5,7 @@
 namespace Thesis\Nats\JetStream\KeyValue;
 
 use Thesis\Nats\Header;
+use Thesis\Nats\Headers;
 
 /**
  * @api
@@ -16,15 +17,37 @@ enum EntryState
     case Purged;
 
     /**
+     * The operation is taken from the KV-Operation header of {@see Bucket::delete()} and {@see Bucket::purge()},
+     * or from the Nats-Marker-Reason header of a marker the server places itself when a key expires or is removed
+     * (see {@see BucketConfig::$limitMarkerTtl}).
+     *
      * @internal
-     * @param ?Header\KvOperation::OP_* $op
      */
-    public static function fromKVOperation(?string $op): self
+    public static function fromHeaders(?Headers $headers): self
     {
-        return match ($op) {
-            Header\KvOperation::OP_DEL => EntryState::Deleted,
-            Header\KvOperation::OP_PURGE => EntryState::Purged,
-            default => EntryState::Created,
+        if ($headers === null) {
+            return self::Created;
+        }
+
+        $op = $headers->get(Header\KvOperation::header());
+
+        if ($op !== null) {
+            return match ($op) {
+                Header\KvOperation::OP_DEL => self::Deleted,
+                Header\KvOperation::OP_PURGE => self::Purged,
+                default => self::Created,
+            };
+        }
+
+        return match ($headers->get(Header\MarkerReason::header())) {
+            'MaxAge', 'Purge' => self::Purged,
+            'Remove' => self::Deleted,
+            default => self::Created,
         };
     }
+
+    public function isMarker(): bool
+    {
+        return $this !== self::Created;
+    }
 }
diff --git a/src/JetStream/KeyValue/Internal/EntryCollector.php b/src/JetStream/KeyValue/Internal/EntryCollector.php
new file mode 100644
index 0000000..48119bd
--- /dev/null
+++ b/src/JetStream/KeyValue/Internal/EntryCollector.php
@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\JetStream\KeyValue\Internal;
+
+use Thesis\Nats\JetStream\KeyValue\Entry;
+
+/**
+ * A watch handler that keeps the entries, for the operations that read the initial values of a watch and stop it.
+ *
+ * @internal
+ */
+final class EntryCollector
+{
+    /** @var list<Entry> */
+    private array $entries = [];
+
+    public function __invoke(Entry $entry): void
+    {
+        $this->entries[] = $entry;
+    }
+
+    /**
+     * @return list<Entry>
+     */
+    public function entries(): array
+    {
+        return $this->entries;
+    }
+}
diff --git a/src/JetStream/KeyValue/Internal/InitialValues.php b/src/JetStream/KeyValue/Internal/InitialValues.php
new file mode 100644
index 0000000..bfc4e73
--- /dev/null
+++ b/src/JetStream/KeyValue/Internal/InitialValues.php
@@ -0,0 +1,77 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\JetStream\KeyValue\Internal;
+
+use Amp\DeferredFuture;
+use Amp\Future;
+
+/**
+ * Tracks the values a watch has to deliver before it is up to date: as many as the consumer had pending when it was created.
+ * Like nats.go, the count is cut short by a message with nothing pending after it, e.g. when some of the values were removed meanwhile,
+ * and unlike nats.go, by an idle heartbeat, when the values were removed before any was delivered.
+ *
+ * @internal
+ */
+final class InitialValues
+{
+    /** @var DeferredFuture<void> */
+    private readonly DeferredFuture $delivered;
+
+    private int $received = 0;
+
+    /**
+     * @param non-negative-int $pending
+     */
+    public function __construct(
+        private readonly int $pending,
+    ) {
+        /** @var DeferredFuture<void> $delivered */
+        $delivered = new DeferredFuture();
+        $this->delivered = $delivered;
+
+        if ($this->pending === 0) {
+            $this->delivered->complete();
+        }
+    }
+
+    public static function none(): self
+    {
+        return new self(0);
+    }
+
+    /**
+     * @param non-negative-int $pendingAfter the number of messages the consumer has pending after the received one
+     */
+    public function receive(int $pendingAfter): void
+    {
+        if ($this->delivered()) {
+            return;
+        }
+
+        if (++$this->received >= $this->pending || $pendingAfter === 0) {
+            $this->delivered->complete();
+        }
+    }
+
+    public function complete(): void
+    {
+        if (!$this->delivered()) {
+            $this->delivered->complete();
+        }
+    }
+
+    public function delivered(): bool
+    {
+        return $this->delivered->isComplete();
+    }
+
+    /**
+     * @return Future<void>
+     */
+    public function future(): Future
+    {
+        return $this->delivered->getFuture();
+    }
+}
diff --git a/src/JetStream/KeyValue/Internal/WatchHandler.php b/src/JetStream/KeyValue/Internal/WatchHandler.php
new file mode 100644
index 0000000..ed9cc96
--- /dev/null
+++ b/src/JetStream/KeyValue/Internal/WatchHandler.php
@@ -0,0 +1,83 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\JetStream\KeyValue\Internal;
+
+use Thesis\Nats\Header;
+use Thesis\Nats\JetStream\Delivery;
+use Thesis\Nats\JetStream\Internal\IdleListener;
+use Thesis\Nats\JetStream\KeyValue\Entry;
+use Thesis\Nats\JetStream\KeyValue\EntryState;
+use Thesis\Nats\Subscription;
+
+/**
+ * Turns the messages of a watch consumer into entries and counts the initial values.
+ *
+ * @internal
+ */
+final readonly class WatchHandler implements IdleListener
+{
+    /**
+     * @param callable(Entry, Subscription): void $handler
+     * @param non-empty-string $bucket
+     * @param non-empty-string $prefix
+     * @param bool $initialValuesOnly drop the updates that come after the initial values, for the reads that stop the watch then
+     */
+    public function __construct(
+        private mixed $handler,
+        private string $bucket,
+        private string $prefix,
+        private bool $ignoreDeletes,
+        private InitialValues $initialValues,
+        private bool $initialValuesOnly = false,
+    ) {}
+
+    public function __invoke(Delivery $delivery, Subscription $subscription): void
+    {
+        $metadata = $delivery->metadata;
+        if ($metadata === null || ($this->initialValuesOnly && $this->initialValues->delivered())) {
+            return;
+        }
+
+        $key = substr($delivery->subject, \strlen($this->prefix));
+        if ($key === '') {
+            return;
+        }
+
+        $pending = max($metadata->pending, 0);
+        $state = EntryState::fromHeaders($delivery->message->headers);
+
+        if (!$this->ignoreDeletes || !$state->isMarker()) {
+            $payload = $delivery->message->payload;
+
+            ($this->handler)(
+                new Entry(
+                    bucket: $this->bucket,
+                    key: $key,
+                    created: $metadata->timestamp,
+                    revision: max($metadata->streamSequence, 0),
+                    size: $payload !== null
+                        ? \strlen($payload)
+                        : $delivery->message->headers?->get(Header\MsgSize::header()) ?? 0,
+                    value: $payload,
+                    delta: $pending,
+                    state: $state,
+                ),
+                $subscription,
+            );
+        }
+
+        // The entry is handled first, so that the initial values are complete once they are reported to be.
+        $this->initialValues->receive($pending);
+    }
+
+    /**
+     * The server has nothing more to deliver, so the initial values are all here even if fewer came than were pending,
+     * e.g. when the bucket was purged after the watch had started.
+     */
+    public function idle(): void
+    {
+        $this->initialValues->complete();
+    }
+}
diff --git a/src/JetStream/KeyValue/WatchConfig.php b/src/JetStream/KeyValue/WatchConfig.php
index d06c01f..3c7f662 100644
--- a/src/JetStream/KeyValue/WatchConfig.php
+++ b/src/JetStream/KeyValue/WatchConfig.php
@@ -4,13 +4,44 @@
 
 namespace Thesis\Nats\JetStream\KeyValue;
 
+use Thesis\Nats\Internal\Validation\Validate;
+
 /**
+ * By default a watch first delivers the latest entry of every watched key (deletes included),
+ * {@see Watcher::awaitInitialValues()} returns once they all have been handled, and then the watch delivers the updates.
+ *
  * @api
  */
 final readonly class WatchConfig
 {
+    /**
+     * @param bool $headersOnly deprecated, use $metaOnly
+     * @param bool $ignoreDeletes do not pass delete and purge markers to the handler
+     * @param bool $includeHistory deliver every stored revision of the watched keys, not only the latest ones
+     * @param bool $updatesOnly deliver only the changes made after the watch has started, without the current values
+     * @param bool $metaOnly deliver entries without values, {@see Entry::$size} still has the size of the value
+     * @param ?positive-int $resumeFromRevision deliver every revision starting from this one, e.g. to continue a watch that was stopped
+     * @throws \InvalidArgumentException
+     */
     public function __construct(
+        /** @deprecated use {@see WatchConfig::$metaOnly} */
         public bool $headersOnly = false,
         public bool $ignoreDeletes = false,
-    ) {}
+        public bool $includeHistory = false,
+        public bool $updatesOnly = false,
+        public bool $metaOnly = false,
+        public ?int $resumeFromRevision = null,
+    ) {
+        if ($this->includeHistory && $this->updatesOnly) {
+            throw new \InvalidArgumentException('Include history can not be used with updates only.');
+        }
+
+        if ($this->updatesOnly && $this->resumeFromRevision !== null) {
+            throw new \InvalidArgumentException('Resume from revision can not be used with updates only.');
+        }
+
+        if ($this->resumeFromRevision !== null) {
+            Validate::count($this->resumeFromRevision, 'revision to resume from');
+        }
+    }
 }
diff --git a/src/JetStream/KeyValue/Watcher.php b/src/JetStream/KeyValue/Watcher.php
new file mode 100644
index 0000000..df601fb
--- /dev/null
+++ b/src/JetStream/KeyValue/Watcher.php
@@ -0,0 +1,98 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\JetStream\KeyValue;
+
+use Amp\Cancellation;
+use Amp\CancelledException;
+use Amp\Future;
+use Thesis\Nats\Exception\ConsumerNotFound;
+use Thesis\Nats\Exception\StreamNotFound;
+use Thesis\Nats\JetStream\Consumer;
+use Thesis\Nats\NatsException;
+use Thesis\Nats\Subscription;
+
+/**
+ * A running {@see Bucket::watch()}: the subscription that passes entries to the handler and the consumer behind it.
+ *
+ * @api
+ */
+final readonly class Watcher
+{
+    /**
+     * @internal
+     * @param Future<void> $initialValues
+     * @param Future<void> $completion
+     */
+    public function __construct(
+        private Subscription $subscription,
+        private Consumer $consumer,
+        private Future $initialValues,
+        private Future $completion,
+    ) {}
+
+    /**
+     * Waits until the handler has been called for every value the bucket had when the watch started
+     * (for every stored revision with {@see WatchConfig::$includeHistory}, from the revision of {@see WatchConfig::$resumeFromRevision}).
+     * Returns at once if there were none or with {@see WatchConfig::$updatesOnly}, and when the watch completes before that;
+     * if the watch fails, e.g. the handler throws, the error is rethrown.
+     */
+    public function awaitInitialValues(?Cancellation $cancellation = null): void
+    {
+        Future\awaitFirst([$this->initialValues, $this->completion], $cancellation);
+    }
+
+    /**
+     * Whether the handler has been called for every initial value, see {@see awaitInitialValues()}.
+     */
+    public function initialValuesDelivered(): bool
+    {
+        return $this->initialValues->isComplete();
+    }
+
+    /**
+     * Stops the subscription, the received entries that are not handled yet are dropped, and deletes the consumer.
+     *
+     * @throws CancelledException
+     * @throws NatsException
+     */
+    public function stop(?Cancellation $cancellation = null): void
+    {
+        $this->subscription->stop($cancellation);
+        $this->deleteConsumer($cancellation);
+    }
+
+    /**
+     * Stops the subscription after the received entries are handled and deletes the consumer.
+     *
+     * @throws NatsException
+     */
+    public function drain(?Cancellation $cancellation = null): void
+    {
+        $this->subscription->drain($cancellation);
+        $this->deleteConsumer($cancellation);
+    }
+
+    public function awaitCompletion(?Cancellation $cancellation = null): void
+    {
+        $this->subscription->awaitCompletion($cancellation);
+    }
+
+    public function completed(): bool
+    {
+        return $this->subscription->completed();
+    }
+
+    /**
+     * @throws NatsException
+     */
+    private function deleteConsumer(?Cancellation $cancellation): void
+    {
+        try {
+            $this->consumer->delete($cancellation);
+        } catch (ConsumerNotFound|StreamNotFound) {
+            // Already gone, e.g. with the bucket.
+        }
+    }
+}
diff --git a/src/JetStream/Stream.php b/src/JetStream/Stream.php
index 20f7467..9d3a277 100644
--- a/src/JetStream/Stream.php
+++ b/src/JetStream/Stream.php
@@ -4,6 +4,7 @@
 
 namespace Thesis\Nats\JetStream;
 
+use Amp\Cancellation;
 use Thesis\Nats\Client;
 use Thesis\Nats\Description;
 use Thesis\Nats\Exception\HeadersIsInvalid;
@@ -57,12 +58,14 @@ public function purge(
         ?int $sequence = null,
         ?int $keep = null,
         ?string $subject = null,
+        ?Cancellation $cancellation = null,
     ): Api\StreamPurged {
         return $this->js->purgeStream(
             name: $this->name,
             sequence: $sequence,
             keep: $keep,
             subject: $subject,
+            cancellation: $cancellation,
         );
     }
 
@@ -112,18 +115,20 @@ public function updateConsumer(Api\ConsumerConfig $config = new Api\ConsumerConf
     /**
      * @throws NatsException
      */
-    public function createOrUpdateConsumer(Api\ConsumerConfig $config = new Api\ConsumerConfig()): Consumer
-    {
-        return $this->js->createOrUpdateConsumer($this->name, $config);
+    public function createOrUpdateConsumer(
+        Api\ConsumerConfig $config = new Api\ConsumerConfig(),
+        ?Cancellation $cancellation = null,
+    ): Consumer {
+        return $this->js->createOrUpdateConsumer($this->name, $config, $cancellation);
     }
 
     /**
      * @param non-empty-string $consumer
      * @throws NatsException
      */
-    public function deleteConsumer(string $consumer): Api\ConsumerDeleted
+    public function deleteConsumer(string $consumer, ?Cancellation $cancellation = null): Api\ConsumerDeleted
     {
-        return $this->js->deleteConsumer($this->name, $consumer);
+        return $this->js->deleteConsumer($this->name, $consumer, $cancellation);
     }
 
     /**
diff --git a/tests/JetStream/KeyValue/KeyValueTest.php b/tests/JetStream/KeyValue/KeyValueTest.php
index 511f056..40d1553 100644
--- a/tests/JetStream/KeyValue/KeyValueTest.php
+++ b/tests/JetStream/KeyValue/KeyValueTest.php
@@ -4,14 +4,112 @@
 
 namespace Thesis\Nats\JetStream\KeyValue;
 
+use Amp\CancelledException;
+use Amp\TimeoutCancellation;
 use PHPUnit\Framework\Attributes\CoversClass;
+use Revolt\EventLoop;
+use Thesis\Nats\Client;
+use Thesis\Nats\Exception\StreamNameAlreadyInUse;
+use Thesis\Nats\Exception\StreamNotFound;
+use Thesis\Nats\Exception\WrongLastSequence;
+use Thesis\Nats\JetStream;
+use Thesis\Nats\JetStream\Api\StoreCompression;
+use Thesis\Nats\JetStream\Api\StreamSource;
+use Thesis\Nats\JetStream\Api\SubjectTransformConfig;
 use Thesis\Nats\NatsTestCase;
+use Thesis\Nats\PausableProxy;
 use Thesis\Nats\Subscription;
+use Thesis\Time\TimeSpan;
+use function Amp\async;
+use function Amp\delay;
 use function Thesis\Nats\Internal\Id\generateUniqueId;
 
 #[CoversClass(Bucket::class)]
+#[CoversClass(BucketStatus::class)]
+#[CoversClass(EntryState::class)]
+#[CoversClass(WatchConfig::class)]
+#[CoversClass(Watcher::class)]
+#[CoversClass(JetStream::class)]
 final class KeyValueTest extends NatsTestCase
 {
+    private const float TIMEOUT = 5;
+
+    private ?PausableProxy $proxy = null;
+
+    #[\Override]
+    protected function tearDown(): void
+    {
+        parent::tearDown();
+        $this->proxy?->stop();
+    }
+
+    public function testInitialValuesAreDoneWhenBucketIsPurgedBeforeDelivery(): void
+    {
+        $this->proxy = PausableProxy::start($this->natsAddress());
+        $js = $this->clientWithUrls([$this->proxy->address])->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+        $kv->put('x', '1');
+        $kv->put('y', '2');
+
+        // The inbox of the API replies is subscribed already, so the next SUB is the one of the watch:
+        // the consumer counts two pending values, but the server does not deliver them until the SUB arrives.
+        $held = $this->proxy->pauseBefore('SUB ');
+        $keys = async($kv->keys(...));
+        $held->await(new TimeoutCancellation(self::TIMEOUT));
+
+        $other = new Client($this->config());
+
+        try {
+            $other->jetStream()->purgeStream("KV_{$bucket}");
+        } finally {
+            $other->stop();
+        }
+
+        $this->proxy->resume();
+
+        // Nothing is delivered then, the idle heartbeat of the watch tells that there is nothing more.
+        self::assertSame([], $keys->await(new TimeoutCancellation(self::TIMEOUT * 2)));
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testCancellationLimitsReadsOfWatch(): void
+    {
+        $this->proxy = PausableProxy::start($this->natsAddress());
+        $js = $this->clientWithUrls([$this->proxy->address])->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+        $kv->put('x', '1');
+
+        // The server does not see the requests of the reads, so only the cancellation can end them.
+        $this->proxy->pause();
+        $resume = EventLoop::delay(self::TIMEOUT, $this->proxy->resume(...));
+
+        foreach ([
+            static fn(): array => $kv->keys(cancellation: new TimeoutCancellation(0.1)),
+            static fn(): array => $kv->history('x', cancellation: new TimeoutCancellation(0.1)),
+            static fn() => $kv->purgeDeletes(cancellation: new TimeoutCancellation(0.1)),
+        ] as $read) {
+            $start = hrtime(true);
+
+            try {
+                $read();
+                self::fail('The server does not see the request, the read must be cancelled.');
+            } catch (CancelledException) {
+            }
+
+            self::assertLessThan(1, (hrtime(true) - $start) / 1e9);
+        }
+
+        EventLoop::cancel($resume);
+        $this->proxy->resume();
+
+        self::assertSame(['x'], $kv->keys());
+
+        $js->deleteKeyValue($bucket);
+    }
+
     public function testCreateKeyValue(): void
     {
         $js = $this->client()->jetStream();
@@ -45,7 +143,7 @@ public function testPutGetKeyValue(): void
 
         $entry = $kv->get('x');
         self::assertNotNull($entry);
-        self::assertSame("KV_{$kv->name}", $entry->bucket);
+        self::assertSame($kv->name, $entry->bucket);
         self::assertSame('x', $entry->key);
         self::assertEquals('y', $entry->value);
         self::assertSame(1, $entry->revision);
@@ -357,4 +455,676 @@ static function (Entry $entry, Subscription $subscription) use (
 
         $js->deleteKeyValue($bucket);
     }
+
+    public function testCreate(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        self::assertSame(1, $kv->create('x', 'y'));
+
+        try {
+            $kv->create('x', 'z');
+            self::fail('The second create of the key is expected to fail.');
+        } catch (WrongLastSequence $e) {
+            self::assertSame(10071, $e->errCode);
+        }
+
+        self::assertSame('y', $kv->get('x')?->value);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testCreateOverDeletedKey(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        $kv->put('x', 'y');
+        $kv->delete('x');
+
+        self::assertSame(3, $kv->create('x', 'z'));
+
+        $entry = $kv->get('x');
+        self::assertNotNull($entry);
+        self::assertSame(EntryState::Created, $entry->state);
+        self::assertSame('z', $entry->value);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testCreateOverPurgedKey(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 5));
+
+        $kv->put('x', 'y');
+        $kv->put('x', 'w');
+        $kv->purge('x');
+
+        self::assertSame(4, $kv->create('x', 'z'));
+        self::assertSame('z', $kv->get('x')?->value);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testPutWithTtlAndCreateOverExpiredKey(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig(
+            bucket: $bucket = generateUniqueId(10),
+            limitMarkerTtl: TimeSpan::fromSeconds(5),
+        ));
+
+        $kv->put('x', 'y', ttl: TimeSpan::fromSeconds(1));
+        self::assertSame('y', $kv->get('x')?->value);
+
+        // Once the entry expires, the server places a marker with the Nats-Marker-Reason header.
+        $deadline = microtime(true) + self::TIMEOUT;
+
+        do {
+            delay(0.1);
+            $state = $kv->get('x')?->state;
+        } while ($state === EntryState::Created && microtime(true) < $deadline);
+
+        self::assertSame(EntryState::Purged, $state);
+
+        $revision = $kv->create('x', 'z', ttl: TimeSpan::fromSeconds(60));
+
+        $entry = $kv->get('x');
+        self::assertNotNull($entry);
+        self::assertSame($revision, $entry->revision);
+        self::assertSame('z', $entry->value);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testHistory(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 10));
+
+        $kv->put('x', '1');
+        $kv->put('a', 'b');
+        $kv->put('x', '2');
+        $kv->delete('x');
+
+        self::assertSame(
+            [
+                [1, '1', EntryState::Created],
+                [3, '2', EntryState::Created],
+                [4, null, EntryState::Deleted],
+            ],
+            array_map(
+                static fn(Entry $entry): array => [$entry->revision, $entry->value, $entry->state],
+                $kv->history('x'),
+            ),
+        );
+
+        self::assertSame([], $kv->history('unknown'));
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testHistoryOfKeysMatchingPattern(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 10));
+
+        $kv->put('app.env', 'prod');
+        $kv->put('database.dsn', 'mysql:');
+        $kv->put('app.debug', '1');
+        $kv->put('app.env', 'dev');
+
+        self::assertSame(
+            [
+                [1, 'app.env', 'prod'],
+                [3, 'app.debug', '1'],
+                [4, 'app.env', 'dev'],
+            ],
+            array_map(
+                static fn(Entry $entry): array => [$entry->revision, $entry->key, $entry->value],
+                $kv->history('app.*'),
+            ),
+        );
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testKeys(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 5));
+
+        self::assertSame([], $kv->keys());
+
+        $kv->put('app.env', 'prod');
+        $kv->put('app.debug', '0');
+        $kv->put('app.env', 'dev');
+        $kv->put('db.dsn', 'mysql://');
+        $kv->put('db.user', 'root');
+        $kv->delete('db.user');
+        $kv->put('cache.dsn', 'redis://');
+        $kv->purge('cache.dsn');
+
+        $keys = $kv->keys();
+        sort($keys);
+        self::assertSame(['app.debug', 'app.env', 'db.dsn'], $keys);
+
+        $keys = $kv->keys('app.*');
+        sort($keys);
+        self::assertSame(['app.debug', 'app.env'], $keys);
+
+        $keys = $kv->keys(['app.env', 'db.>']);
+        sort($keys);
+        self::assertSame(['app.env', 'db.dsn'], $keys);
+
+        // The watch consumers of the reads are deleted once they are done.
+        self::assertSame([], [...$js->consumerNames("KV_{$bucket}")]);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testStatus(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig(
+            bucket: $bucket = generateUniqueId(10),
+            history: 3,
+            ttl: TimeSpan::fromHours(1),
+            compression: true,
+        ));
+
+        $kv->put('x', 'y');
+        $kv->put('x', 'z');
+
+        $status = $kv->status();
+        self::assertSame($bucket, $status->bucket);
+        self::assertSame(2, $status->values);
+        self::assertSame(3, $status->history);
+        self::assertSame(3600, $status->ttl?->toSeconds());
+        self::assertGreaterThan(0, $status->bytes);
+        self::assertSame('JetStream', $status->backingStore);
+        self::assertTrue($status->compressed);
+        self::assertNull($status->limitMarkerTtl);
+        self::assertSame(StoreCompression::S2, $status->info->config->compression);
+
+        $plain = $js->createOrUpdateKeyValue(new BucketConfig($plainBucket = generateUniqueId(10)))->status();
+        self::assertSame(0, $plain->values);
+        self::assertNull($plain->ttl);
+        self::assertFalse($plain->compressed);
+
+        $js->deleteKeyValue($bucket);
+        $js->deleteKeyValue($plainBucket);
+    }
+
+    public function testPurgeDeletesKeepsRecentMarkers(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 10));
+
+        $kv->put('x', '1');
+        $kv->put('x', '2');
+        $kv->delete('x');
+        $kv->put('y', '1');
+        $kv->purge('y');
+        $kv->put('z', '1');
+
+        $kv->purgeDeletes();
+
+        self::assertSame(
+            [EntryState::Deleted],
+            array_map(static fn(Entry $entry): EntryState => $entry->state, $kv->history('x')),
+        );
+        self::assertSame(
+            [EntryState::Purged],
+            array_map(static fn(Entry $entry): EntryState => $entry->state, $kv->history('y')),
+        );
+        self::assertSame('1', $kv->get('z')?->value);
+        self::assertSame(3, $kv->status()->values);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testPurgeDeletesRemovesOldMarkers(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 10));
+
+        $kv->put('x', '1');
+        $kv->delete('x');
+        $kv->put('y', '1');
+        $kv->purge('y');
+        $kv->put('z', '1');
+
+        $kv->purgeDeletes(TimeSpan::fromSeconds(0));
+
+        self::assertSame([], $kv->history('x'));
+        self::assertSame([], $kv->history('y'));
+        self::assertNull($kv->get('x'));
+        self::assertSame(['z'], $kv->keys());
+        self::assertSame(1, $kv->status()->values);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testCreateKeyValueOnly(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+        $kv->put('x', 'y');
+
+        // The same config is not a conflict.
+        self::assertSame('y', $js->createKeyValue(new BucketConfig($bucket))->get('x')?->value);
+
+        try {
+            $js->createKeyValue(new BucketConfig($bucket, history: 5));
+            self::fail('Creating an existing bucket with another config is expected to fail.');
+        } catch (StreamNameAlreadyInUse) {
+        }
+
+        self::assertSame(1, $kv->status()->history);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testUpdateKeyValueOnly(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $bucket = generateUniqueId(10);
+
+        try {
+            $js->updateKeyValue(new BucketConfig($bucket));
+            self::fail('Updating a missing bucket is expected to fail.');
+        } catch (StreamNotFound) {
+        }
+
+        self::assertNull($js->keyValue($bucket));
+
+        $js->createKeyValue(new BucketConfig($bucket));
+        $kv = $js->updateKeyValue(new BucketConfig($bucket, history: 5));
+        self::assertSame(5, $kv->status()->history);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testMirror(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $origin = $js->createOrUpdateKeyValue(new BucketConfig($originBucket = generateUniqueId(10)));
+        $origin->put('name', 'derek');
+        $origin->put('age', '22');
+
+        $mirror = $js->createOrUpdateKeyValue(new BucketConfig(
+            bucket: $mirrorBucket = generateUniqueId(10),
+            mirror: new StreamSource($originBucket),
+        ));
+
+        $config = $mirror->status()->info->config;
+        self::assertSame("KV_{$originBucket}", $config->mirror?->name);
+        self::assertTrue($config->mirrorDirect);
+        self::assertSame([], $config->subjects ?? []);
+
+        self::assertSame('derek', self::awaitValue($mirror, 'name')?->value);
+        self::assertSame('22', self::awaitValue($mirror, 'age')?->value);
+
+        // A put to the mirror goes to the origin bucket and comes back to the mirror.
+        $mirror->put('name', 'rip');
+        self::assertSame('rip', $origin->get('name')?->value);
+        self::assertSame('rip', self::awaitValue($mirror, 'name', 'rip')?->value);
+
+        // The bucket bound by name is a mirror as well.
+        $bound = $js->keyValue($mirrorBucket);
+        self::assertNotNull($bound);
+        self::assertSame('rip', $bound->get('name')?->value);
+
+        $keys = $mirror->keys();
+        sort($keys);
+        self::assertSame(['age', 'name'], $keys);
+
+        $js->deleteKeyValue($mirrorBucket);
+        $js->deleteKeyValue($originBucket);
+    }
+
+    public function testSources(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $first = $js->createOrUpdateKeyValue(new BucketConfig($firstBucket = generateUniqueId(10)));
+        $second = $js->createOrUpdateKeyValue(new BucketConfig($secondBucket = generateUniqueId(10)));
+
+        $sourced = $js->createOrUpdateKeyValue(new BucketConfig(
+            bucket: $sourcedBucket = generateUniqueId(10),
+            sources: [
+                new StreamSource($firstBucket),
+                new StreamSource("KV_{$secondBucket}"),
+            ],
+        ));
+
+        $first->put('key1', 'value1');
+        $second->put('key2', 'value2');
+
+        self::assertSame('value1', self::awaitValue($sourced, 'key1')?->value);
+        self::assertSame('value2', self::awaitValue($sourced, 'key2')?->value);
+
+        // Own entries live next to the sourced ones.
+        $sourced->put('key3', 'value3');
+        self::assertSame('value3', $sourced->get('key3')?->value);
+
+        $config = $sourced->status()->info->config;
+        self::assertSame(["\$KV.{$sourcedBucket}.>"], $config->subjects);
+        self::assertEquals(
+            [
+                [new SubjectTransformConfig("\$KV.{$firstBucket}.>", "\$KV.{$sourcedBucket}.>")],
+                [new SubjectTransformConfig("\$KV.{$secondBucket}.>", "\$KV.{$sourcedBucket}.>")],
+            ],
+            array_map(static fn(StreamSource $source): ?array => $source->subjectTransforms, $config->sources ?? []),
+        );
+        self::assertSame(
+            ["KV_{$firstBucket}", "KV_{$secondBucket}"],
+            array_map(static fn(StreamSource $source): string => $source->name, $config->sources ?? []),
+        );
+
+        $js->deleteKeyValue($sourcedBucket);
+        $js->deleteKeyValue($firstBucket);
+        $js->deleteKeyValue($secondBucket);
+    }
+
+    public function testWatchDeliversLatestValuesFirst(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 5));
+
+        $kv->put('x', '1');
+        $kv->put('y', '2');
+        $kv->put('x', '3');
+        $kv->put('z', '4');
+        $kv->delete('z');
+
+        /** @var \ArrayObject<int, array{string, int, ?string, EntryState}> $entries */
+        $entries = new \ArrayObject();
+
+        $watcher = $kv->watch(static function (Entry $entry) use ($entries): void {
+            $entries[] = [$entry->key, $entry->revision, $entry->value, $entry->state];
+        });
+
+        $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+        self::assertTrue($watcher->initialValuesDelivered());
+
+        self::assertSame(
+            [
+                ['y', 2, '2', EntryState::Created],
+                ['x', 3, '3', EntryState::Created],
+                ['z', 5, null, EntryState::Deleted],
+            ],
+            $entries->getArrayCopy(),
+        );
+
+        $kv->put('y', '6');
+        self::awaitCount($entries, 4);
+        self::assertSame(['y', 6, '6', EntryState::Created], $entries[3]);
+
+        $watcher->stop();
+        self::assertTrue($watcher->completed());
+        self::assertSame([], [...$js->consumerNames("KV_{$bucket}")]);
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchEmptyBucketHasNoInitialValues(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        /** @var \ArrayObject<int, string> $keys */
+        $keys = new \ArrayObject();
+
+        $watcher = $kv->watch(static function (Entry $entry) use ($keys): void {
+            $keys[] = $entry->key;
+        });
+
+        self::assertTrue($watcher->initialValuesDelivered());
+        $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+
+        $kv->put('x', 'y');
+        self::awaitCount($keys, 1);
+        self::assertSame(['x'], $keys->getArrayCopy());
+
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchIncludeHistory(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 5));
+
+        $kv->put('x', '1');
+        $kv->put('y', '2');
+        $kv->put('x', '3');
+
+        /** @var \ArrayObject<int, string> $entries */
+        $entries = new \ArrayObject();
+
+        $watcher = $kv->watch(
+            static function (Entry $entry) use ($entries): void {
+                $entries[] = "{$entry->key}:{$entry->revision}";
+            },
+            keys: 'x',
+            config: new WatchConfig(includeHistory: true),
+        );
+
+        $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+        self::assertSame(['x:1', 'x:3'], $entries->getArrayCopy());
+
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchUpdatesOnly(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        $kv->put('x', '1');
+
+        /** @var \ArrayObject<int, string> $entries */
+        $entries = new \ArrayObject();
+
+        $watcher = $kv->watch(
+            static function (Entry $entry) use ($entries): void {
+                $entries[] = "{$entry->key}:{$entry->value}";
+            },
+            config: new WatchConfig(updatesOnly: true),
+        );
+
+        self::assertTrue($watcher->initialValuesDelivered());
+
+        $kv->put('y', '2');
+        self::awaitCount($entries, 1);
+        self::assertSame(['y:2'], $entries->getArrayCopy());
+
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchMetaOnly(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        $kv->put('x', 'yyy');
+
+        /** @var \ArrayObject<int, array{string, ?string, int}> $entries */
+        $entries = new \ArrayObject();
+
+        $watcher = $kv->watch(
+            static function (Entry $entry) use ($entries): void {
+                $entries[] = [$entry->key, $entry->value, $entry->size];
+            },
+            config: new WatchConfig(metaOnly: true),
+        );
+
+        $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+        self::assertSame([['x', null, 3]], $entries->getArrayCopy());
+
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchResumeFromRevision(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10), history: 5));
+
+        $kv->put('x', '1');
+        $kv->put('y', '2');
+        $kv->put('x', '3');
+        $kv->put('z', '4');
+
+        /** @var \ArrayObject<int, string> $entries */
+        $entries = new \ArrayObject();
+
+        $watcher = $kv->watch(
+            static function (Entry $entry) use ($entries): void {
+                $entries[] = "{$entry->key}:{$entry->revision}";
+            },
+            config: new WatchConfig(resumeFromRevision: 2),
+        );
+
+        $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+        self::assertSame(['y:2', 'x:3', 'z:4'], $entries->getArrayCopy());
+
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchIgnoreDeletesCountsSkippedInitialValues(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        $kv->put('x', '1');
+        $kv->put('y', '2');
+        $kv->delete('y');
+
+        /** @var \ArrayObject<int, string> $entries */
+        $entries = new \ArrayObject();
+
+        $watcher = $kv->watch(
+            static function (Entry $entry) use ($entries): void {
+                $entries[] = $entry->key;
+            },
+            config: new WatchConfig(ignoreDeletes: true),
+        );
+
+        $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+        self::assertSame(['x'], $entries->getArrayCopy());
+
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testAwaitInitialValuesRethrowsHandlerError(): void
+    {
+        $js = $this->client()->jetStream();
+
+        $kv = $js->createOrUpdateKeyValue(new BucketConfig($bucket = generateUniqueId(10)));
+
+        $kv->put('x', '1');
+        $kv->put('y', '2');
+
+        $watcher = $kv->watch(static function (Entry $entry): void {
+            throw new \RuntimeException("Cannot handle {$entry->key}.");
+        });
+
+        try {
+            $watcher->awaitInitialValues(new TimeoutCancellation(self::TIMEOUT));
+            self::fail('The error of the handler is expected.');
+        } catch (\RuntimeException $e) {
+            self::assertSame('Cannot handle x.', $e->getMessage());
+        }
+
+        self::assertTrue($watcher->completed());
+        $watcher->stop();
+
+        $js->deleteKeyValue($bucket);
+    }
+
+    public function testWatchConfigRejectsConflictingOptions(): void
+    {
+        foreach ([
+            static fn(): WatchConfig => new WatchConfig(includeHistory: true, updatesOnly: true),
+            static fn(): WatchConfig => new WatchConfig(updatesOnly: true, resumeFromRevision: 1),
+            /** @phpstan-ignore argument.type */
+            static fn(): WatchConfig => new WatchConfig(resumeFromRevision: 0),
+        ] as $config) {
+            try {
+                $config();
+                self::fail('The config is expected to be rejected.');
+            } catch (\InvalidArgumentException) {
+            }
+        }
+
+        self::assertTrue((new WatchConfig(includeHistory: true, resumeFromRevision: 1))->includeHistory);
+    }
+
+    /**
+     * @param non-empty-string $key
+     */
+    private static function awaitValue(Bucket $kv, string $key, ?string $value = null): ?Entry
+    {
+        $deadline = microtime(true) + self::TIMEOUT;
+
+        do {
+            $entry = $kv->get($key);
+
+            if ($entry !== null && ($value === null || $entry->value === $value)) {
+                return $entry;
+            }
+
+            delay(0.05);
+        } while (microtime(true) < $deadline);
+
+        return $entry;
+    }
+
+    /**
+     * @template T
+     * @param \ArrayObject<int, T> $entries
+     */
+    private static function awaitCount(\ArrayObject $entries, int $count): void
+    {
+        $deadline = microtime(true) + self::TIMEOUT;
+
+        while (\count($entries) < $count && microtime(true) < $deadline) {
+            delay(0.01);
+        }
+    }
 }
diff --git a/tests/JetStreamTest.php b/tests/JetStreamTest.php
index 6359667..1dc79dc 100644
--- a/tests/JetStreamTest.php
+++ b/tests/JetStreamTest.php
@@ -74,6 +74,38 @@ public function testCreateStream(): void
         $stream->delete();
     }
 
+    public function testStreamInfoHasConfigLimits(): void
+    {
+        $client = $this->client();
+        $js = $client->jetStream();
+
+        $subject = generateUniqueId(10);
+
+        $stream = $js->createStream(new StreamConfig(
+            name: generateUniqueId(10),
+            subjects: [$subject],
+            maxMessages: 100,
+            maxMessagesPerSubject: 7,
+            maxMessageSize: 1024,
+            storageType: JetStream\Api\StorageType::Memory,
+            allowRollup: true,
+            rePublish: new JetStream\Api\RePublish(dest: "republished.{$subject}", src: $subject, headersOnly: true),
+            allowMessageTtl: true,
+        ));
+
+        $config = $js->streamInfo($stream->name)->config;
+        self::assertSame(100, $config->maxMessages);
+        self::assertSame(7, $config->maxMessagesPerSubject);
+        self::assertSame(1024, $config->maxMessageSize);
+        self::assertSame(JetStream\Api\StorageType::Memory, $config->storageType);
+        self::assertSame(1, $config->replicas);
+        self::assertTrue($config->allowRollup);
+        self::assertTrue($config->allowMessageTtl);
+        self::assertEquals(new JetStream\Api\RePublish(dest: "republished.{$subject}", src: $subject, headersOnly: true), $config->rePublish);
+
+        $stream->delete();
+    }
+
     public function testUpdateStream(): void
     {
         $client = $this->client();
diff --git a/tests/PausableProxy.php b/tests/PausableProxy.php
index 01a9763..a03f17e 100644
--- a/tests/PausableProxy.php
+++ b/tests/PausableProxy.php
@@ -5,6 +5,7 @@
 namespace Thesis\Nats;
 
 use Amp\DeferredFuture;
+use Amp\Future;
 use Amp\Socket\ServerSocket;
 use Amp\Socket\Socket;
 use Revolt\EventLoop;
@@ -22,6 +23,9 @@ final class PausableProxy
     /** @var ?DeferredFuture<void> */
     private ?DeferredFuture $paused = null;
 
+    /** @var ?array{non-empty-string, DeferredFuture<void>} */
+    private ?array $pauseBefore = null;
+
     /**
      * @param non-empty-string $address
      */
@@ -67,6 +71,21 @@ public function pause(): void
         $this->paused ??= new DeferredFuture();
     }
 
+    /**
+     * Holds back what the clients send starting from the given bytes, until resume().
+     *
+     * @param non-empty-string $needle
+     * @return Future<void> completes once the bytes have been sent and held back
+     */
+    public function pauseBefore(string $needle): Future
+    {
+        /** @var DeferredFuture<void> $held */
+        $held = new DeferredFuture();
+        $this->pauseBefore = [$needle, $held];
+
+        return $held->getFuture();
+    }
+
     /**
      * Forwards what has been held back and everything the clients send next.
      */
@@ -86,6 +105,17 @@ private function forward(Socket $from, Socket $to, bool $pausable): void
     {
         try {
             while (($chunk = $from->read()) !== null) {
+                if ($pausable && $this->pauseBefore !== null && ($position = strpos($chunk, $this->pauseBefore[0])) !== false) {
+                    [, $held] = $this->pauseBefore;
+                    $this->pauseBefore = null;
+
+                    $to->write(substr($chunk, 0, $position));
+                    $chunk = substr($chunk, $position);
+
+                    $this->pause();
+                    $held->complete();
+                }
+
                 while ($pausable && ($paused = $this->paused) !== null) {
                     $paused->getFuture()->await();
                 }