#ifndef HYPERSWARM_OVERLAY_TRANSPORT_H
#define HYPERSWARM_OVERLAY_TRANSPORT_H
// Binds liboverlay's abstract Transport to this project's Protomux.
//
// WHY THIS FILE IS HERE AND NOT IN liboverlay/. The dependency points this way on
// purpose: liboverlay knows nothing about Protomux, and the adapter knows about
// both. That is what keeps liboverlay liftable — extraction stays a directory
// move, and whoever lifts it writes their own 60-line adapter for whatever
// carries their frames. Putting this inside liboverlay would have dragged
// hyperswarm-peer/src/ along with it.
//
// THE ONE INVARIANT THAT MATTERS: messages are registered in SLOT ORDER. Protomux
// numbers messages by registration order on both sides, so the Nth add_message
// here must be the Nth on the peer. Register them in a different order — or skip
// one — and every frame lands in the wrong handler on a peer that did it right,
// with no error anywhere. MsgId IS that order, and the static_asserts below pin
// it so a reordering of the enum breaks the build rather than the wire.
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>
#include <memory>
#include "debug_log.h"
#include "protomux.h"
#include "overlay/protocol.h"
#include "overlay/transport.h"
namespace hyperswarm {
// Channel name must match the JS exactly (lib/protocol.js PROTOCOL) or the two
// never pair.
static constexpr const char *OVERLAY_PROTOCOL = "hyper-overlay/v2";
class OverlayTransport : public hyperoverlay::Transport {
public:
// `mux` and `proto` must outlive this object. The channel is created here so
// slot registration cannot drift from the enum.
// CHANNEL CREATION IS NOT ONE-SHOT, and that is the whole point of this
// constructor's shape.
//
// If the remote has not opened its overlay channel yet — it booted a moment
// later, or is mid-restart — create_channel returns null, and a transport that
// gave up there would never carry a byte for the LIFE of that connection while
// everything else on the mux kept working perfectly. The JS reference documents
// exactly this failure (lib/protocol.js:241-259: "the pointers keep converging
// while the BYTES never move again") and its fix is mux.pair(), which fires
// when the remote asks for a protocol we have no channel for.
//
// So: try now, and register a pair callback that tries again when the remote
// turns up. The callback stays registered after a close, so a peer that comes
// back gets a channel too.
//
// `ready_cb` IS A CONSTRUCTOR ARGUMENT AND NOT A FIELD TO ASSIGN AFTERWARDS,
// because a channel can reach `opened` before this constructor returns: if the
// remote's Open is already waiting, create_channel pairs and fully opens it
// synchronously. A listener installed on the next line is then too late to hear
// about a connection that is working, and the peer sits outside the fetch pool.
// The JS reference passes onopen into createChannel for the same reason.
OverlayTransport(Protomux *mux, hyperoverlay::Protocol *proto,
const std::string &label = "peer",
std::function<void(OverlayTransport *)> ready_cb = nullptr)
: on_ready(std::move(ready_cb)), mux_(mux), proto_(proto), label_(label) {
// THE CALLBACK MUST NOT OUTLIVE `this`, and it otherwise does.
//
// Protomux stores this lambda in its _notify map and is owned by the
// PeerConnection, which outlives us: HyperOverlay::udx_remove_peer()
// deletes the transport when a peer drops, and the slot-mismatch path
// deletes it immediately after construction. A raw `this` capture would
// then leave the mux holding a dangling pointer, and the next pair()
// event would read channel_ and run open_channel() through freed memory —
// writing to it and registering fourteen more lambdas that also capture
// the dead object. The corruption surfaces later and elsewhere, as
// "free(): double free detected in tcache" (hyper-games #18).
//
// A destructor calling unpair() would fix the common case but depends on
// the mux still being alive at that moment, which is exactly the ordering
// we cannot check through a raw pointer. The liveness token has no such
// hazard: an expired weak_ptr means the transport is gone and the
// callback is a no-op, whichever object died first. A replacement
// transport on the same mux re-registers under the same key, so stale
// entries are overwritten rather than accumulated.
std::weak_ptr<char> alive = alive_;
mux->pair(OVERLAY_PROTOCOL, [this, alive](const std::string &, const std::vector<uint8_t> &) {
if (alive.expired()) return; // transport destroyed; nothing to open
DHT_LOG("[overlay] pair() fired for %s (channel=%s)\n",
label_.c_str(), channel_ ? "already-open" : "creating");
if (!channel_) open_channel();
});
open_channel();
}
// True once the channel is OPEN, not merely created. The distinction is the
// whole bug this shape fixes: ProtomuxChannel::send refuses on its first line
// while `opened` is false, so a peer added at creation time has every frame
// dropped before it reaches the wire — silently, with no error anywhere.
bool ready() const { return channel_ != nullptr && channel_->opened; }
// Whether this transport is holding a channel at all. A closed one is FORGOTTEN
// (see open_channel), so this reads false again after a reject and the pair
// callback is free to build another.
bool has_channel() const { return channel_ != nullptr; }
// Fires once per channel, when it is open and carrying frames — whether that
// happened during construction, on a deferred pairing, or on a channel built to
// replace one the remote rejected. Set it through the constructor: a channel
// can already be open by the time this object exists.
std::function<void(OverlayTransport *)> on_ready;
private:
// Tell the owner the channel carries frames now — once per channel. Called from
// two places because a channel can reach `opened` two ways (see open_channel),
// and the owner adding the same peer twice is not the kind of thing to leave to
// Protocol::add_peer being idempotent today.
void notify_ready() {
if (ready_notified_) return;
ready_notified_ = true;
if (on_ready) on_ready(this);
}
void open_channel() {
channel_ = mux_->create_channel(OVERLAY_PROTOCOL);
DHT_LOG("[overlay] create_channel(%s) for %s -> %s\n",
OVERLAY_PROTOCOL, label_.c_str(), channel_ ? "OK" : "NULL");
if (!channel_) return;
// Registration order IS the wire contract — see the header comment.
for (uint8_t i = 0; i < (uint8_t)hyperoverlay::MsgId::COUNT; i++) {
const hyperoverlay::MsgId id = (hyperoverlay::MsgId)i;
const uint32_t slot = channel_->add_message(
[this, id](const uint8_t *d, size_t n, ProtomuxChannel *) {
// No per-frame trace here on purpose: this fires once per chunkData,
// which is the flood debug_log.h exists to keep out of the default
// build. What arrives is observable from the fetch's outcome and
// from the four lifecycle lines in protocol.h (HO_VERBOSE=1).
if (!proto_) return;
std::vector<uint8_t> frame(d, d + n);
// `this` IS the peer, as far as the protocol is concerned: one
// OverlayTransport per connection. The driver keys per-peer state
// off it, so a multi-peer fetch can tell its holders apart.
proto_->on_message(this, id, frame);
});
// If Protomux ever hands back a slot that is not the registration
// index, the assumption above is wrong and everything downstream is
// silently misrouted — so notice it here rather than in the field.
if (slot != i) slot_mismatch_ = true;
}
// The carrier drained: resume anything the serve pump parked. This is the
// return leg of the writability signal — send() reports the buffer is full,
// this reports it is not any more.
//
// One drain callback per mux, and the overlay creates one channel per
// connection, so this does not contend. A second consumer on the same mux
// would need Protomux to fan the signal out instead.
mux_->set_drain_callback([this]() {
if (proto_) proto_->peer_writable(this);
});
// Opening is ASYNCHRONOUS: it completes when the remote announces its
// matching channel. Anything sent in between is discarded, so the peer must
// not join the fetch pool until this fires. udx_transport does exactly this
// for its godot-rpc channel — the pattern was already in the repo.
//
// Both hooks take the liveness token for the reason pair() does: the channel
// is owned by the mux, which OUTLIVES this transport, and Protomux::shutdown
// closes every channel it still holds. A raw `this` would then run
// on_close through freed memory during teardown.
std::weak_ptr<char> alive = alive_;
channel_->on_open = [this, alive](const uint8_t *, size_t) {
if (alive.expired()) return;
DHT_LOG("[overlay] channel OPEN for %s\n", label_.c_str());
notify_ready();
};
// A CLOSED CHANNEL MUST BE FORGOTTEN, or the pair callback above declines to
// build the next one and this connection never carries a byte again.
//
// A peer that joins a mesh before its overlay attaches has no handler for
// the protocol, so protomux rejects our Open and _on_reject closes this
// channel. When that peer attaches a moment later and opens its own, pair()
// fires here exactly as designed — and used to find `channel_` still
// pointing at the rejected channel, so `if (!channel_)` was false and
// nothing re-opened. The remote's Open then sat unpaired for the life of the
// connection: layer 13's third peer, meshed and attached and unable to fetch.
//
// The JS reference has no state to go stale — hyper-overlay keys its peer
// entry by mux and DELETES it in onclose, so the pair callback's re-entry
// into attach(mux) always builds a fresh channel. This is that deletion.
//
// The peer leaves the pool at the same moment, which is the other half of
// what onclose does there: its inflight chunks go back so a survivor can
// take them, rather than being owed by a channel that is gone.
channel_->on_close = [this, alive](bool) {
if (alive.expired()) return;
DHT_LOG("[overlay] channel CLOSED for %s — forgetting it\n", label_.c_str());
channel_ = nullptr;
ready_notified_ = false;
if (proto_) proto_->remove_peer(this);
};
channel_->open();
// create_channel FULLY OPENS the channel itself when the remote's Open is
// already waiting — synchronously, before this function could register
// anything, so on_open never fires for it and the owner would never hear
// about a channel that is working. (The JS reference cannot hit this: it
// takes the handlers as createChannel arguments and defers the open by a
// tick.) That is the ordinary case on the side that re-opens for a late
// attacher, which is to say: on the only peer that can serve it.
if (channel_ && channel_->opened) notify_ready();
}
public:
bool send(hyperoverlay::MsgId id, const std::vector<uint8_t> &frame) override {
if (!channel_) {
// A real fault, not trace: the remote never paired, so this frame is
// discarded. Kept behind the gate because a peer that has not opened its
// overlay channel yet is the NORMAL case during a mesh forming, and this
// would otherwise fire once per frame for every such peer.
DHT_LOG("[overlay] SEND %s DROPPED on %s — no channel\n",
hyperoverlay::msg_name(id), label_.c_str());
return false;
}
// Protomux's own answer, forwarded rather than discarded — see
// overlay/transport.h for why that return value exists at all.
return channel_->send((uint32_t)id, frame.data(), frame.size());
}
const char *peer_label() const override { return label_.c_str(); }
bool ok() const { return channel_ != nullptr && !slot_mismatch_; }
bool slot_mismatch() const { return slot_mismatch_; }
ProtomuxChannel *channel() const { return channel_; }
private:
Protomux *mux_ = nullptr;
ProtomuxChannel *channel_ = nullptr;
hyperoverlay::Protocol *proto_ = nullptr;
std::string label_;
bool slot_mismatch_ = false;
// Per-channel, not per-transport: cleared on close so the next channel on this
// same connection announces itself too.
bool ready_notified_ = false;
// Liveness token for the pair() callback held by the mux. Its only job is to
// die with this object, so a callback that outlives us can tell. Declared
// last, destroyed first among these — the weak_ptr expires the moment
// destruction begins.
std::shared_ptr<char> alive_ = std::make_shared<char>(0);
};
// The enum is the wire order; a reorder must not compile.
static_assert((uint8_t)hyperoverlay::MsgId::SyncState == 0, "overlay slot order changed");
static_assert((uint8_t)hyperoverlay::MsgId::ChunkHashes == 3, "overlay slot order changed");
static_assert((uint8_t)hyperoverlay::MsgId::ChunkData == 5, "overlay slot order changed");
static_assert((uint8_t)hyperoverlay::MsgId::ContentRequest == 11, "overlay slot order changed");
static_assert((uint8_t)hyperoverlay::MsgId::COUNT == 14, "overlay message count changed");
} // namespace hyperswarm
#endif // HYPERSWARM_OVERLAY_TRANSPORT_H