NAME
EV::Nats::ObjectStore - Chunked object store on top of NATS JetStream
SYNOPSIS
use EV;
use EV::Nats;
use EV::Nats::JetStream;
use EV::Nats::ObjectStore;
my $nats = EV::Nats->new(host => '127.0.0.1');
my $js = EV::Nats::JetStream->new(nats => $nats);
my $os = EV::Nats::ObjectStore->new(js => $js, bucket => 'files');
$os->create_bucket({}, sub {
$os->put('report.pdf', $pdf_data, sub {
my ($info, $err) = @_;
print "stored: $info->{size} bytes in $info->{chunks} chunks\n";
$os->get('report.pdf', sub {
my ($data, $err, $meta) = @_;
print "got $meta->{size} bytes back\n";
});
});
});
EV::run;DESCRIPTION
An object-store bucket is a JetStream stream named OBJ_<bucket> with two subject groups:
$O.<bucket>.C.<nuid>- opaque chunks for one object (one chunk per stream message; the nuid is generated per object).$O.<bucket>.M.<encoded-name>- last-write-wins metadata describing an object (name, size, chunk count, SHA-256 digest).
put chunks the input, publishes each chunk via js_publish for durability, then writes a metadata entry. get fetches the metadata, walks the chunks back via STREAM.MSG.GET, and verifies the digest.
METHODS
All callbacks fire on the EV loop, not synchronously.
new
new(js => $js, bucket => $name, [chunk_size => $bytes])Default chunk_size is 128 KiB. timeout defaults to the timeout of $js.
create_bucket
create_bucket(\%opts, $cb)Provision the underlying stream. Recognised \%opts:
max_bytes- bucket-wide storage cap.max_age- per-message TTL in nanoseconds.replicas- cluster replication factor.
Callback: ($info, $err).
delete_bucket
delete_bucket($cb)Tear down the underlying stream. Callback: ($info, $err).
put
put($name, $data, $cb)Store $data under $name, automatically chunked. Each chunk is published with JetStream ack; the metadata entry is written last (with a Nats-Rollup: sub header, so the meta subject holds exactly one message) so a partial upload doesn't surface a half-stored object. Overwriting an existing name purges the previous object's chunks after the new metadata is acknowledged. Callback: ($info, $err) where $info is { name, size, chunks, seq }.
get
get($name, $cb)Retrieve a previously-stored object. Callback: ($data, $err, $meta). $data is undef if the object does not exist or has been deleted (both the old KV-Operation tombstone and the nats.go "deleted":true marker are recognised). On digest mismatch, $data is undef and $err is "digest mismatch". Digests written by 0.03/0.04 (hex), 0.05 (unpadded base64url) and 0.06+/nats.go (padded base64url) all verify.
delete
delete($name, [$cb])Publishes a "deleted":true metadata marker with a Nats-Rollup: sub header (the nats.go form; the PubAck doubles as the flush fence), then best-effort purges the object's chunks under $O.<bucket>.C.<nuid>. Marking is done first, so a purge hiccup cannot leave a "deleted" object still readable. Idempotent: deleting a missing or already-deleted object is a no-op success and writes no marker. Callback: ($ok, $err).
info
info($name, $cb)Fetch only the metadata entry for an object, without downloading chunks. Callback: (\%meta, $err); \%meta is undef if the object does not exist or was deleted (the deletion marker is recognised). This is the recommended way to filter live objects out of a "list" result.
list
list($cb)List object names in the bucket. Callback: (\@names, $err). Names written by 0.03-0.05 (legacy %XX encoding) and by 0.06+/nats.go (base64url) are both decoded. Deleted entries still appear in the listing -- filtering them would cost a per-name metadata round-trip, so call info to filter.
status
status($cb)Returns a snapshot hashref:
{ bucket => $name, bytes => $n, sealed => 0|1 }sealed reflects the underlying stream's config.sealed flag; this client never seals on its own, so unless someone manually sealed the stream out-of-band the value is always 0.
Callback: (\%status, $err).
SEE ALSO
EV::Nats, EV::Nats::JetStream, EV::Nats::KV, NATS Object Store.