feat: add handleTtlOp to server
This commit is contained in:
1 parent
a283b01556
commit
d577007961
1 file changed
+149
+149
@@ -470,3 +470,152 @@ pub const Server = struct {
|
|||||||
// unconditionally rather than ever consulting a session store.
|
// unconditionally rather than ever consulting a session store.
|
||||||
.auth_token => try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""),
|
.auth_token => try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""),
|
||||||
|
|
||||||
|
.get => {
|
||||||
|
const value = self.db.get(self.allocator, frame.payload) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
if (value) |v| {
|
||||||
|
defer self.allocator.free(v);
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), v);
|
||||||
|
} else {
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.not_found), "");
|
||||||
|
}
|
||||||
|
},
|
||||||
|
|
||||||
|
.set => {
|
||||||
|
const args = protocol.decodeSet(frame.payload) catch {
|
||||||
|
return writeErr(writer, .bad_request, "malformed set payload");
|
||||||
|
};
|
||||||
|
self.db.put(args.key, args.value) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
|
||||||
|
},
|
||||||
|
|
||||||
|
.del => {
|
||||||
|
self.db.delete(frame.payload) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
|
||||||
|
},
|
||||||
|
|
||||||
|
.scan => try self.handleScan(frame.payload, writer),
|
||||||
|
.prefix => try self.handlePrefix(frame.payload, writer),
|
||||||
|
|
||||||
|
.stats => {
|
||||||
|
const s = self.db.stats();
|
||||||
|
const payload = protocol.encodeStats(self.allocator, s) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
defer self.allocator.free(payload);
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), payload);
|
||||||
|
},
|
||||||
|
|
||||||
|
.flush => {
|
||||||
|
self.db.flush() catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
|
||||||
|
},
|
||||||
|
|
||||||
|
.compact => {
|
||||||
|
self.db.compact() catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
|
||||||
|
},
|
||||||
|
|
||||||
|
.batch => try self.handleBatch(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_select => try self.handleTableSelect(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_join => try self.handleTableJoin(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_aggregate => try self.handleTableAggregate(frame.payload, writer),
|
||||||
|
.table_window => try self.handleTableWindow(frame.payload, writer),
|
||||||
|
|
||||||
|
// All three already handled and returned from above, before
|
||||||
|
// `self.mutex` was ever taken - never reached here.
|
||||||
|
.subscribe, .publish, .unsubscribe => unreachable,
|
||||||
|
|
||||||
|
.table_explain => try self.handleTableExplain(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_update => try self.handleTableUpdate(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_delete => try self.handleTableDelete(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_export => try self.handleTableExport(frame.payload, writer),
|
||||||
|
|
||||||
|
.table_import => try self.handleTableImport(frame.payload, writer),
|
||||||
|
|
||||||
|
.user_admin => try self.handleUserAdmin(frame.payload, writer),
|
||||||
|
|
||||||
|
.metrics => try self.handleMetrics(writer),
|
||||||
|
|
||||||
|
.ttl_op => try self.handleTtlOp(frame.payload, writer),
|
||||||
|
|
||||||
|
.coll_ttl_op => try self.handleCollTtlOp(frame.payload, writer),
|
||||||
|
|
||||||
|
.coll_op => try self.handleCollOp(frame.payload, writer),
|
||||||
|
|
||||||
|
.txn_op => try self.handleTxnOp(frame.payload, writer),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Answers the `ttl_op` opcode: the network counterpart of the CLI's
|
||||||
|
/// local `expire`/`persist`/`ttl`/`expire-sweep` (main.zig), all backed
|
||||||
|
/// by the same `ttl.zig` functions the local commands call directly -
|
||||||
|
/// `query`/`sweep` were previously reachable only from a process that
|
||||||
|
/// had the `Db` open itself. `expire` mirrors the CLI's Redis-EXPIRE-
|
||||||
|
/// strict existence check (main.zig checks `db.get` before calling
|
||||||
|
/// `ttl.setExpireSeconds`; here the same check happens before the same
|
||||||
|
/// call) and answers `not_found` rather than `ok` when the key does not
|
||||||
|
/// currently exist, instead of the looser PEXPIREAT-like
|
||||||
|
/// `ttl.setExpireAt` that would silently accept it.
|
||||||
|
fn handleTtlOp(self: *Server, payload: []const u8, writer: anytype) !void {
|
||||||
|
const args = protocol.decodeTtlOp(payload) catch {
|
||||||
|
return writeErr(writer, .bad_request, "malformed ttl payload");
|
||||||
|
};
|
||||||
|
const now = std.time.nanoTimestamp();
|
||||||
|
switch (args.action) {
|
||||||
|
.expire => {
|
||||||
|
const existing = self.db.get(self.allocator, args.key) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
if (existing == null) return writeErr(writer, .not_found, "no such key");
|
||||||
|
self.allocator.free(existing.?);
|
||||||
|
ttlmod.setExpireSeconds(self.db, self.allocator, args.key, now, args.seconds) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
|
||||||
|
},
|
||||||
|
.persist => {
|
||||||
|
ttlmod.clearExpire(self.db, self.allocator, args.key) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
|
||||||
|
},
|
||||||
|
.query => {
|
||||||
|
const result = ttlmod.remaining(self.db, self.allocator, args.key, now) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
const respPayload = protocol.encodeTtlResult(self.allocator, result) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
defer self.allocator.free(respPayload);
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), respPayload);
|
||||||
|
},
|
||||||
|
.sweep => {
|
||||||
|
const reaped = ttlmod.sweepExpired(self.db, self.allocator, now) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
if (reaped > 0) self.ttlSweptTotal.incrBy(@intCast(reaped));
|
||||||
|
const respPayload = protocol.encodeTtlSweepCount(self.allocator, @intCast(reaped)) catch |err| {
|
||||||
|
return writeErr(writer, .internal, @errorName(err));
|
||||||
|
};
|
||||||
|
defer self.allocator.free(respPayload);
|
||||||
|
try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), respPayload);
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user