EV-Etcd
view release on metacpan or search on metacpan
Revision history for EV-Etcd
0.08 2026-06-02
- Fix election observe never delivering events (missing client half-close)
- Validate txn ops before allocation; doc and test fixes
0.07 2026-04-29
- Fix use-after-free on cancelled streaming handles past client cleanup
- Fix memory leak in txn op size validation
- Fix int64/uint64 truncation on 32-bit Perl
- Document response shapes; add ENCODING section; polish POD
- CI: add Linux ARM runners
- Expand t/ coverage; add xt/ author tests and four eg/ patterns
0.06 2026-03-17
- Add opts to lease_keepalive (auto_reconnect) and member_list (linearizable)
- Preserve watch_id across reconnects
- Remove GRPC_STATUS_INTERNAL from retryable set
- Add missing use EV () before XSLoader::load
0.03 2026-03-02
- Fix error sources, election_observe arg order, watch early-cancel
- Clear auth token on auth_disable
- Deduplicate response handlers, remove dead code
0.02 2026-02-10
- Initial release
- KV operations: get, put, delete, range, txn (compare-and-swap)
- Watch with bidirectional streaming and auto-reconnect
- Lease: grant, revoke, keepalive, time-to-live, leases
- Lock and unlock tied to leases
- Election: campaign, proclaim, leader, resign, observe
- Cluster: member list/add/remove/update/promote
- Maintenance: status, compact, defragment, alarm, hash_kv, move_leader
- Auth: user/role management, authenticate, enable/disable
- Health monitoring with configurable interval and callback
- Automatic retries for transient gRPC failures
- Multiple endpoint support with failover
- Structured error callbacks ({code, message, source})
/* Serialize request */
grpc_slice req_slice;
SERIALIZE_PROTOBUF_TO_SLICE(req_slice,
etcdserverpb__watch_request__get_packed_size,
etcdserverpb__watch_request__pack, &req);
if (range_end_copy) Safefree(range_end_copy);
grpc_byte_buffer *send_buffer = grpc_raw_byte_buffer_create(&req_slice, 1);
grpc_slice_unref(req_slice);
/* Create streaming call */
gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME); /* No timeout for watch */
wc->call = grpc_channel_create_call(
client->channel,
NULL, /* parent call */
GRPC_PROPAGATE_DEFAULTS,
client->cq,
METHOD_WATCH,
NULL, /* host */
deadline,
req.id = lease_id;
/* Serialize request */
grpc_slice req_slice;
SERIALIZE_PROTOBUF_TO_SLICE(req_slice,
etcdserverpb__lease_keep_alive_request__get_packed_size,
etcdserverpb__lease_keep_alive_request__pack, &req);
grpc_byte_buffer *send_buffer = grpc_raw_byte_buffer_create(&req_slice, 1);
grpc_slice_unref(req_slice);
/* Create streaming call */
gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
kc->call = grpc_channel_create_call(
client->channel,
NULL,
GRPC_PROPAGATE_DEFAULTS,
client->cq,
METHOD_LEASE_KEEPALIVE,
NULL,
deadline,
req.name.data = (uint8_t *)name_str;
req.name.len = name_len;
grpc_slice req_slice;
SERIALIZE_PROTOBUF_TO_SLICE(req_slice,
v3electionpb__leader_request__get_packed_size,
v3electionpb__leader_request__pack, &req);
grpc_byte_buffer *send_buffer = grpc_raw_byte_buffer_create(&req_slice, 1);
grpc_slice_unref(req_slice);
/* Use infinite deadline for streaming call */
gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
oc->call = grpc_channel_create_call(
client->channel, NULL, GRPC_PROPAGATE_DEFAULTS,
client->cq, METHOD_ELECTION_OBSERVE, NULL, deadline, NULL
);
if (!oc->call) {
grpc_byte_buffer_destroy(send_buffer);
grpc_metadata_array_destroy(&oc->initial_metadata);
ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;
setup_auth_metadata(client, &ops[0], &auth_md);
ops[1].op = GRPC_OP_RECV_INITIAL_METADATA;
ops[1].data.recv_initial_metadata.recv_initial_metadata = &oc->initial_metadata;
ops[2].op = GRPC_OP_SEND_MESSAGE;
ops[2].data.send_message.send_message = send_buffer;
/* Observe is server-streaming: the client sends one LeaderRequest and the
* server streams LeaderResponses. Half-close the client side so etcd's
* handler receives the completed request and begins streaming; without it
* the stream is established but never delivers an event. (Watch/keepalive
* are bidi and deliberately stay open, so they must NOT half-close.) */
ops[3].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
ops[4].op = GRPC_OP_RECV_MESSAGE;
ops[4].data.recv_message.recv_message = &oc->recv_buffer;
grpc_call_error err = grpc_call_start_batch(oc->call, ops, 5, &oc->base, NULL);
cleanup_auth_metadata(client, &auth_md);
grpc_byte_buffer_destroy(send_buffer);
client->active = 0;
/* Stop ev_async watcher */
if (ev_is_active(&client->cq_async)) {
ev_async_stop(EV_DEFAULT, &client->cq_async);
}
/* Signal the gRPC thread to stop and wait for it */
client->thread_running = 0;
/* Mark every streaming call inactive and cancel its gRPC call so pending
* batches complete promptly with success=0. */
watch_call_t *wc = client->watches;
while (wc) {
wc->active = 0;
if (ev_is_active(&wc->reconnect_timer))
ev_timer_stop(EV_DEFAULT, &wc->reconnect_timer);
if (wc->call) {
grpc_call_cancel(wc->call, NULL);
}
wc = wc->next;
t/kv.t
t/kv_advanced.t
t/lease.t
t/limits.t
t/lock.t
t/maintenance.t
t/move_leader.t
t/parameters.t
t/reconnect_drop.t
t/retry_config.t
t/streaming.t
t/txn.t
t/txn_range.t
t/watch_prev_kv.t
t/watch_progress.t
t/watch_reconnect.t
t/watch_resume.t
typemap
xt/changes.t
xt/kwalitee.t
xt/leak.t
say "$event->{type} on $event->{kv}{key}";
}
});
EV::run;
```
## Features
- **KV**: get, put, delete, range, transactions (compare-and-swap)
- **Watch**: bidirectional streaming with auto-reconnect
- **Lease**: grant, revoke, keepalive, time-to-live
- **Lock**: distributed locking tied to leases
- **Election**: leader campaign, observe, proclaim, resign
- **Cluster**: member list/add/remove/update/promote
- **Maintenance**: status, compact, defragment, alarm, hash_kv, move_leader
- **Auth**: user/role management, authenticate, enable/disable
- **Health monitoring** with configurable interval and callback
- **Automatic retries** for transient gRPC failures
## Architecture
eg/resumable_watch.pl view on Meta::CPAN
#!/usr/bin/env perl
#
# resumable_watch.pl - A watcher that survives process restarts without
# missing events. Persists the last seen revision to a file; on startup,
# reads it back, lists any keys changed during downtime via a one-shot
# get(prefix=>1, revision=>) and then resumes streaming from the next
# revision.
#
# Handles the compaction edge case: if the server has compacted past our
# saved revision, the watch arrives in the *error* callback (per the POD
# documentation of watch's error semantics). We then re-list at HEAD and
# accept the gap.
#
# Try it:
# $ perl eg/resumable_watch.pl /myapp/config/
# $ etcdctl put /myapp/config/foo bar
etcd_common.h view on Meta::CPAN
size_t key_len;
char *range_end;
size_t range_end_len;
int64_t start_revision;
int prev_kv;
int progress_notify;
int64_t watch_id;
int has_watch_id;
} watch_params_t;
/* Watch structure (for streaming watch) */
typedef struct watch_call {
call_base_t base; /* Must be first */
grpc_call *call;
SV *callback;
grpc_metadata_array initial_metadata;
grpc_metadata_array trailing_metadata;
grpc_byte_buffer *recv_buffer;
grpc_slice status_details;
int64_t watch_id;
int active;
etcd_common.h view on Meta::CPAN
watch_params_t params;
int reconnect_attempt;
ev_timer reconnect_timer; /* Backoff timer for reconnection */
/* Dual ownership: gRPC state freed by client-side cleanup, struct freed by
* the last owner. Prevents use-after-free when Perl holds the handle past
* client cleanup. */
int client_owns;
int perl_owns;
} watch_call_t;
/* Keepalive structure (for streaming lease keepalive) */
typedef struct keepalive_call {
call_base_t base; /* Must be first */
grpc_call *call;
SV *callback;
grpc_metadata_array initial_metadata;
grpc_metadata_array trailing_metadata;
grpc_byte_buffer *recv_buffer;
grpc_slice status_details;
int64_t lease_id;
int active;
etcd_common.h view on Meta::CPAN
int client_owns; /* See watch_call_t — same dual-ownership */
int perl_owns;
} keepalive_call_t;
/* Election observe parameters for reconnection */
typedef struct observe_params {
char *name;
size_t name_len;
} observe_params_t;
/* Election observe structure (for streaming election observe) */
typedef struct observe_call {
call_base_t base; /* Must be first */
grpc_call *call;
SV *callback;
grpc_metadata_array initial_metadata;
grpc_metadata_array trailing_metadata;
grpc_byte_buffer *recv_buffer;
grpc_slice status_details;
int active;
struct ev_etcd_struct *client;
etcd_common.h view on Meta::CPAN
grpc_metadata_array_destroy(&(pc)->initial_metadata); \
grpc_metadata_array_destroy(&(pc)->trailing_metadata); \
if ((pc)->recv_buffer) grpc_byte_buffer_destroy((pc)->recv_buffer); \
grpc_slice_unref((pc)->status_details); \
if ((pc)->call) grpc_call_unref((pc)->call); \
SvREFCNT_dec((pc)->callback); \
Safefree((pc)); \
} while (0)
/*
* Helper macros for streaming call reconnection to reduce code triplication
* across watch, keepalive, and observe reconnect functions.
*/
/*
* Cleanup old streaming call state before reconnection.
* Works with any streaming call struct that has these fields.
*
* Usage:
* STREAMING_CALL_CLEANUP(wc); // For watch_call_t
* STREAMING_CALL_CLEANUP(kc); // For keepalive_call_t
* STREAMING_CALL_CLEANUP(oc); // For observe_call_t
*/
#define STREAMING_CALL_CLEANUP(call_ptr) \
do { \
if ((call_ptr)->call) { \
grpc_call_unref((call_ptr)->call); \
etcd_common.h view on Meta::CPAN
grpc_metadata_array_destroy(&(call_ptr)->initial_metadata); \
grpc_metadata_array_destroy(&(call_ptr)->trailing_metadata); \
if ((call_ptr)->recv_buffer) { \
grpc_byte_buffer_destroy((call_ptr)->recv_buffer); \
(call_ptr)->recv_buffer = NULL; \
} \
grpc_slice_unref((call_ptr)->status_details); \
} while (0)
/*
* Reinitialize streaming call state for reconnection.
*
* Usage:
* STREAMING_CALL_REINIT(wc);
*/
#define STREAMING_CALL_REINIT(call_ptr) \
do { \
grpc_metadata_array_init(&(call_ptr)->initial_metadata); \
grpc_metadata_array_init(&(call_ptr)->trailing_metadata); \
(call_ptr)->status_details = grpc_empty_slice(); \
(call_ptr)->active = 1; \
} while (0)
/*
* Setup standard 4-op batch for streaming call reconnection.
* Requires: ops[4], auth_md, send_buffer, call_ptr all in scope.
*
* Usage:
* STREAMING_CALL_SETUP_OPS(client, ops, auth_md, send_buffer, wc);
*/
#define STREAMING_CALL_SETUP_OPS(client, ops, auth_md, send_buf, call_ptr) \
do { \
(ops)[0].op = GRPC_OP_SEND_INITIAL_METADATA; \
setup_auth_metadata(client, &(ops)[0], &(auth_md)); \
(ops)[1].op = GRPC_OP_RECV_INITIAL_METADATA; \
(ops)[1].data.recv_initial_metadata.recv_initial_metadata = &(call_ptr)->initial_metadata; \
(ops)[2].op = GRPC_OP_SEND_MESSAGE; \
(ops)[2].data.send_message.send_message = (send_buf); \
(ops)[3].op = GRPC_OP_RECV_MESSAGE; \
(ops)[3].data.recv_message.recv_message = &(call_ptr)->recv_buffer; \
} while (0)
/*
* Handle error after failed batch start for streaming reconnect.
* Self-contained: also tears down the metadata arrays and status_details that
* STREAMING_CALL_REINIT just set up, so callers don't need follow-up cleanup
* to avoid leaks.
*
* Usage:
* STREAMING_CALL_BATCH_ERROR(wc);
*/
#define STREAMING_CALL_BATCH_ERROR(call_ptr) \
do { \
(call_ptr)->active = 0; \
etcd_election.c view on Meta::CPAN
(void)revents;
observe_call_t *oc = (observe_call_t *)((char *)w - offsetof(observe_call_t, reconnect_timer));
ev_etcd_t *client = oc->client;
if (!client->active) {
cleanup_observe(aTHX_ oc);
return;
}
/* Cleanup and reinitialize streaming state */
STREAMING_CALL_CLEANUP(oc);
STREAMING_CALL_REINIT(oc);
/* Create LeaderRequest for observe */
V3electionpb__LeaderRequest req = V3ELECTIONPB__LEADER_REQUEST__INIT;
req.name.data = (uint8_t *)oc->params.name;
req.name.len = oc->params.name_len;
grpc_slice req_slice;
SERIALIZE_PROTOBUF_TO_SLICE(req_slice,
etcd_election.c view on Meta::CPAN
return;
}
cleanup_observe(aTHX_ oc);
return;
}
grpc_op ops[5] = {0};
grpc_metadata auth_md;
STREAMING_CALL_SETUP_OPS(client, ops, auth_md, send_buffer, oc);
/* Observe is server-streaming: the shared macro lays out bidi ops (no
* half-close, RECV_MESSAGE at ops[3]). Replace ops[3] with the client
* half-close and move RECV_MESSAGE to ops[4], matching the initial setup,
* so the reconnected stream actually delivers events. */
ops[3].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
ops[4].op = GRPC_OP_RECV_MESSAGE;
ops[4].data.recv_message.recv_message = &oc->recv_buffer;
init_call_base(&oc->base, CALL_TYPE_ELECTION_OBSERVE);
grpc_call_error err = grpc_call_start_batch(oc->call, ops, 5, &oc->base, NULL);
cleanup_auth_metadata(client, &auth_md);
etcd_election.h view on Meta::CPAN
#include "etcd_common.h"
#include "election.pb-c.h"
/* Election response handlers */
void process_campaign_response(pTHX_ pending_call_t *pc);
void process_proclaim_response(pTHX_ pending_call_t *pc);
void process_leader_response(pTHX_ pending_call_t *pc);
void process_resign_response(pTHX_ pending_call_t *pc);
/* Election observe (streaming) handlers */
void process_observe_response(pTHX_ observe_call_t *oc);
void observe_rearm_recv(pTHX_ observe_call_t *oc);
void cleanup_observe(pTHX_ observe_call_t *oc);
void observe_call_perl_release(pTHX_ observe_call_t *oc);
int try_reconnect_observe(pTHX_ observe_call_t *oc);
/* Helper to convert LeaderKey to hash */
HV *leader_key_to_hv(pTHX_ V3electionpb__LeaderKey *lk);
#endif /* ETCD_ELECTION_H */
etcd_lease.c view on Meta::CPAN
(void)revents;
keepalive_call_t *kc = (keepalive_call_t *)((char *)w - offsetof(keepalive_call_t, reconnect_timer));
ev_etcd_t *client = kc->client;
if (!client->active) {
cleanup_keepalive(aTHX_ kc);
return;
}
/* Cleanup and reinitialize streaming state */
STREAMING_CALL_CLEANUP(kc);
STREAMING_CALL_REINIT(kc);
/* Build keepalive request */
Etcdserverpb__LeaseKeepAliveRequest keep_req = ETCDSERVERPB__LEASE_KEEP_ALIVE_REQUEST__INIT;
keep_req.id = kc->lease_id;
grpc_slice req_slice;
SERIALIZE_PROTOBUF_TO_SLICE(req_slice,
etcdserverpb__lease_keep_alive_request__get_packed_size,
etcd_watch.c view on Meta::CPAN
(void)revents;
watch_call_t *wc = (watch_call_t *)((char *)w - offsetof(watch_call_t, reconnect_timer));
ev_etcd_t *client = wc->client;
if (!client->active) {
cleanup_watch(aTHX_ wc);
return;
}
/* Cleanup and reinitialize streaming state */
STREAMING_CALL_CLEANUP(wc);
STREAMING_CALL_REINIT(wc);
/* Build watch create request */
Etcdserverpb__WatchCreateRequest create_req = ETCDSERVERPB__WATCH_CREATE_REQUEST__INIT;
create_req.key.data = (uint8_t *)wc->params.key;
create_req.key.len = wc->params.key_len;
if (wc->params.range_end && wc->params.range_end_len > 0) {
create_req.range_end.data = (uint8_t *)wc->params.range_end;
lib/EV/Etcd.pm view on Meta::CPAN
ArrayRef of etcd endpoints (host:port). Optional; defaults to
C<['127.0.0.1:2379']>. When more than one is provided, the client uses the
first endpoint and rotates to subsequent endpoints on connection failure.
=item timeout
RPC timeout in seconds. Default is 30 seconds. Minimum value is 1 second.
=item max_retries
Maximum number of reconnection attempts for streaming operations (watch,
lease_keepalive, election_observe) after a connection failure. Default is 3.
Set to 0 to disable automatic reconnection.
=item health_interval
Interval in seconds for health monitoring. Default is 0 (disabled).
When enabled, the client periodically checks the gRPC channel connectivity
state and calls the on_health_change callback when the connection state changes.
=item on_health_change
lib/EV/Etcd.pm view on Meta::CPAN
$client->lease_revoke($lease_id, $callback);
Revoke a lease. All keys attached to the lease will be deleted. The response
contains C<header> only.
=head2 lease_keepalive
my $keepalive = $client->lease_keepalive($lease_id, $callback);
my $keepalive = $client->lease_keepalive($lease_id, \%opts, $callback);
Keep a lease alive. Creates a bidirectional streaming connection that keeps
the lease refreshed. Returns an C<EV::Etcd::Keepalive> object that can be
used to cancel the keepalive stream:
$keepalive->cancel(sub { my ($resp, $err) = @_; });
Options:
=over 4
=item auto_reconnect
lib/EV/Etcd.pm view on Meta::CPAN
$client->election_resign($leader, sub {
my ($resp, $err) = @_;
say "Resigned from leadership" unless $err;
});
=head2 election_observe
my $observe = $client->election_observe($name, $callback);
my $observe = $client->election_observe($name, \%opts, $callback);
Observe leader changes for an election. This creates a streaming connection
that receives notifications whenever the leader changes. Returns an
C<EV::Etcd::Observe> object that can be used to cancel the observe stream:
$observe->cancel(sub { my ($resp, $err) = @_; });
Arguments:
=over 4
=item name
t/election.t view on Meta::CPAN
if ($err) {
diag("Expected error after resign: $err->{status} - $err->{message}");
}
EV::break;
});
my $t7 = EV::timer(5, 0, sub { fail('timeout'); EV::break });
EV::run;
}
}
# Test 17-22: election_observe streaming test
{
my $observe_election = "observe-test-$$-" . time();
my $observe_lease_id;
my @observed_events;
my $observe_leader_key;
# Grant a lease for this test
$client->lease_grant(30, sub {
my ($resp, $err) = @_;
$observe_lease_id = $resp->{id} if !$err;
t/streaming.t view on Meta::CPAN
};
plan skip_all => 'etcd not available on 127.0.0.1:2379' unless $etcd_available;
plan tests => 11;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
my $test_prefix = "/test-streaming-$$-" . time();
# === lease_keepalive streaming tests ===
# Test 1-4: lease_keepalive receives response
my $lease_id;
my $keepalive_count = 0;
$client->lease_grant(10, sub {
my ($resp, $err) = @_;
ok(!$err, 'lease_grant succeeded');
$lease_id = $resp->{id};
diag("Granted lease: id=$lease_id, ttl=$resp->{ttl}");
t/streaming.t view on Meta::CPAN
ok($keepalive_count >= 1, "received at least 1 keepalive response (got $keepalive_count)");
# Cleanup lease (this will end the keepalive)
$client->lease_revoke($lease_id, sub {
my ($resp, $err) = @_;
diag($err ? "Revoke failed" : "Lease revoked");
});
pass('cleanup initiated');
}
# === Watch streaming tests ===
# Test 5-9: watch receives multiple events
my $watch_key = "$test_prefix/watch-stream-test";
my $watch_count = 0;
my $watch_target = 3;
my $watch_handle;
$watch_handle = $client->watch($watch_key, sub {
my ($resp, $err) = @_;
if ($err) {
t/streaming.t view on Meta::CPAN
});
}
my $watch_timer = EV::timer 10, 0, sub {
diag("Watch timer expired, received $watch_count events");
EV::break;
};
EV::run;
ok($watch_count >= 1, "watch received at least 1 event (got $watch_count)");
cmp_ok($watch_count, '>=', $watch_target, "watch received all $watch_target events (streaming works)");
# Test watch cancel - uses callback-based cancel API
my $cancel_done = 0;
$watch_handle->cancel(sub {
my ($resp, $err) = @_;
$cancel_done = 1;
diag("Watch cancel callback: " . ($err ? "error: $err" : "success"));
EV::break;
});
my $cancel_timer = EV::timer 5, 0, sub {
t/watch_reconnect.t view on Meta::CPAN
my ($resp, $err) = @_;
return if $err;
if ($resp->{events} && @{$resp->{events}}) {
push @events, @{$resp->{events}};
EV::break if @events >= 3;
}
});
ok(defined $watch, 'watch with default auto_reconnect created');
# Send multiple events to verify streaming works
for my $i (1..3) {
$client->put($key, "value-$i", sub {});
}
my $t3 = EV::timer(5, 0, sub { EV::break });
EV::run;
cmp_ok(scalar @events, '>=', 1, "default watch received events");
# Verify events have expected structure
my $ev = $events[0];
#!/usr/bin/env perl
# Cheap SV-leak smoke test for the dual-ownership streaming-handle path:
# loop new/watch/cancel/DESTROY many times and assert refcounts don't grow.
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use Test::More;
BEGIN { eval { require EV }; plan skip_all => 'EV required' if $@ }
eval "use Devel::Leak; 1" or plan skip_all => 'Devel::Leak required';
use EV;
xt/pod-coverage.t view on Meta::CPAN
#!/usr/bin/env perl
use strict;
use warnings;
use Test::More;
eval "use Test::Pod::Coverage 1.08; 1"
or plan skip_all => 'Test::Pod::Coverage 1.08 required';
eval "use Pod::Coverage 0.18; 1"
or plan skip_all => 'Pod::Coverage 0.18 required';
# Coverage check on EV::Etcd only. The streaming-handle sub-packages
# (EV::Etcd::Watch / Keepalive / Observe) are XS-defined with no .pm of
# their own and only expose cancel + DESTROY — both documented as
# =head3 cancel under their parent service section in lib/EV/Etcd.pm.
# Pod::Coverage can't follow that cross-module reference, so we skip
# them rather than emit false negatives.
pod_coverage_ok(
'EV::Etcd',
{ trustme => [qr/^txn$/] }, # txn is wrapped in pure-Perl, doc'd via XS path
'EV::Etcd has POD coverage',
);
( run in 1.617 second using v1.01-cache-2.11-cpan-995e09ba956 )