feat: add QR scanner, improve UX, and simplify navigation
- Add live camera QR scanner for nsec/ncryptsec login - Replace browser prompt() with proper password dialog for ncryptsec - Add missing /notes/:id route for thread view navigation - Remove explore section entirely (button, page, routes) - Remove profile button from bottom nav, avatar now opens profile - Remove "Notes" tab from feed, default to showing all posts/replies - Add PasswordPromptProvider for secure password input - Add SidebarDrawer for mobile navigation - Add domain layer with value objects and adapters - Various UI and navigation improvements 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
mleku
1275
.claude/skills/nostr-websocket/references/khatru_implementation.md
Normal file
1275
.claude/skills/nostr-websocket/references/khatru_implementation.md
Normal file
File diff suppressed because it is too large
Load Diff
1307
.claude/skills/nostr-websocket/references/rust_implementation.md
Normal file
1307
.claude/skills/nostr-websocket/references/rust_implementation.md
Normal file
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,921 @@
|
||||
# C++ WebSocket Implementation for Nostr Relays (strfry patterns)
|
||||
|
||||
This reference documents high-performance WebSocket patterns from the strfry Nostr relay implementation in C++.
|
||||
|
||||
## Repository Information
|
||||
|
||||
- **Project:** strfry - High-performance Nostr relay
|
||||
- **Repository:** https://github.com/hoytech/strfry
|
||||
- **Language:** C++ (C++20)
|
||||
- **WebSocket Library:** Custom fork of uWebSockets with epoll
|
||||
- **Architecture:** Single-threaded I/O with specialized thread pools
|
||||
|
||||
## Core Architecture
|
||||
|
||||
### Thread Pool Design
|
||||
|
||||
strfry uses 6 specialized thread pools for different operations:
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────────────┐
|
||||
│ Main Thread (I/O) │
|
||||
│ - epoll event loop │
|
||||
│ - WebSocket message reception │
|
||||
│ - Connection management │
|
||||
└─────────────────────────────────────────────────────────────┘
|
||||
│
|
||||
┌───────────────────┼───────────────────┐
|
||||
│ │ │
|
||||
┌────▼────┐ ┌───▼────┐ ┌───▼────┐
|
||||
│Ingester │ │ReqWorker│ │Negentropy│
|
||||
│ (3) │ │ (3) │ │ (2) │
|
||||
└─────────┘ └─────────┘ └─────────┘
|
||||
│ │ │
|
||||
┌────▼────┐ ┌───▼────┐
|
||||
│ Writer │ │ReqMonitor│
|
||||
│ (1) │ │ (3) │
|
||||
└─────────┘ └─────────┘
|
||||
```
|
||||
|
||||
**Thread Pool Responsibilities:**
|
||||
|
||||
1. **WebSocket (1 thread):** Main I/O loop, epoll event handling
|
||||
2. **Ingester (3 threads):** Event validation, signature verification, deduplication
|
||||
3. **Writer (1 thread):** Database writes, event storage
|
||||
4. **ReqWorker (3 threads):** Process REQ subscriptions, query database
|
||||
5. **ReqMonitor (3 threads):** Monitor active subscriptions, send real-time events
|
||||
6. **Negentropy (2 threads):** NIP-77 set reconciliation
|
||||
|
||||
**Deterministic thread assignment:**
|
||||
```cpp
|
||||
int threadId = connId % numThreads;
|
||||
```
|
||||
|
||||
**Benefits:**
|
||||
- **No lock contention:** Shared-nothing architecture
|
||||
- **Predictable performance:** Same connection always same thread
|
||||
- **CPU cache efficiency:** Thread-local data stays hot
|
||||
|
||||
### Connection State
|
||||
|
||||
```cpp
|
||||
struct ConnectionState {
|
||||
uint64_t connId; // Unique connection identifier
|
||||
std::string remoteAddr; // Client IP address
|
||||
|
||||
// Subscription state
|
||||
flat_str subId; // Current subscription ID
|
||||
std::shared_ptr<Subscription> sub; // Subscription filter
|
||||
uint64_t latestEventSent = 0; // Latest event ID sent
|
||||
|
||||
// Compression state (per-message deflate)
|
||||
PerMessageDeflate pmd;
|
||||
|
||||
// Parsing state (reused buffer)
|
||||
std::string parseBuffer;
|
||||
|
||||
// Signature verification context (reused)
|
||||
secp256k1_context *secpCtx;
|
||||
};
|
||||
```
|
||||
|
||||
**Key design decisions:**
|
||||
|
||||
1. **Reusable parseBuffer:** Single allocation per connection
|
||||
2. **Persistent secp256k1_context:** Expensive to create, reused for all signatures
|
||||
3. **Connection ID:** Enables deterministic thread assignment
|
||||
4. **Flat string (flat_str):** Value-semantic string-like type for zero-copy
|
||||
|
||||
## WebSocket Message Reception
|
||||
|
||||
### Main Event Loop (epoll)
|
||||
|
||||
```cpp
|
||||
// Pseudocode representation of strfry's I/O loop
|
||||
uWS::App app;
|
||||
|
||||
app.ws<ConnectionState>("/*", {
|
||||
.compression = uWS::SHARED_COMPRESSOR,
|
||||
.maxPayloadLength = 16 * 1024 * 1024,
|
||||
.idleTimeout = 120,
|
||||
.maxBackpressure = 1 * 1024 * 1024,
|
||||
|
||||
.upgrade = nullptr,
|
||||
|
||||
.open = [](auto *ws) {
|
||||
auto *state = ws->getUserData();
|
||||
state->connId = nextConnId++;
|
||||
state->remoteAddr = getRemoteAddress(ws);
|
||||
state->secpCtx = secp256k1_context_create(SECP256K1_CONTEXT_VERIFY);
|
||||
|
||||
LI << "New connection: " << state->connId << " from " << state->remoteAddr;
|
||||
},
|
||||
|
||||
.message = [](auto *ws, std::string_view message, uWS::OpCode opCode) {
|
||||
auto *state = ws->getUserData();
|
||||
|
||||
// Reuse parseBuffer to avoid allocation
|
||||
state->parseBuffer.assign(message.data(), message.size());
|
||||
|
||||
try {
|
||||
// Parse JSON (nlohmann::json)
|
||||
auto json = nlohmann::json::parse(state->parseBuffer);
|
||||
|
||||
// Extract command type
|
||||
auto cmdStr = json[0].get<std::string>();
|
||||
|
||||
if (cmdStr == "EVENT") {
|
||||
handleEventMessage(ws, std::move(json));
|
||||
}
|
||||
else if (cmdStr == "REQ") {
|
||||
handleReqMessage(ws, std::move(json));
|
||||
}
|
||||
else if (cmdStr == "CLOSE") {
|
||||
handleCloseMessage(ws, std::move(json));
|
||||
}
|
||||
else if (cmdStr == "NEG-OPEN") {
|
||||
handleNegentropyOpen(ws, std::move(json));
|
||||
}
|
||||
else {
|
||||
sendNotice(ws, "unknown command: " + cmdStr);
|
||||
}
|
||||
}
|
||||
catch (std::exception &e) {
|
||||
sendNotice(ws, "Error: " + std::string(e.what()));
|
||||
}
|
||||
},
|
||||
|
||||
.close = [](auto *ws, int code, std::string_view message) {
|
||||
auto *state = ws->getUserData();
|
||||
|
||||
LI << "Connection closed: " << state->connId
|
||||
<< " code=" << code
|
||||
<< " msg=" << std::string(message);
|
||||
|
||||
// Cleanup
|
||||
secp256k1_context_destroy(state->secpCtx);
|
||||
cleanupSubscription(state->connId);
|
||||
},
|
||||
});
|
||||
|
||||
app.listen(8080, [](auto *token) {
|
||||
if (token) {
|
||||
LI << "Listening on port 8080";
|
||||
}
|
||||
});
|
||||
|
||||
app.run();
|
||||
```
|
||||
|
||||
**Key patterns:**
|
||||
|
||||
1. **epoll-based I/O:** Single thread handles thousands of connections
|
||||
2. **Buffer reuse:** `state->parseBuffer` avoids allocation per message
|
||||
3. **Move semantics:** `std::move(json)` transfers ownership to handler
|
||||
4. **Exception handling:** Catches parsing errors, sends NOTICE
|
||||
|
||||
### Message Dispatch to Thread Pools
|
||||
|
||||
```cpp
|
||||
void handleEventMessage(auto *ws, nlohmann::json &&json) {
|
||||
auto *state = ws->getUserData();
|
||||
|
||||
// Pack message with connection ID
|
||||
auto msg = MsgIngester{
|
||||
.connId = state->connId,
|
||||
.payload = std::move(json),
|
||||
};
|
||||
|
||||
// Dispatch to Ingester thread pool (deterministic assignment)
|
||||
tpIngester->dispatchToThread(state->connId, std::move(msg));
|
||||
}
|
||||
|
||||
void handleReqMessage(auto *ws, nlohmann::json &&json) {
|
||||
auto *state = ws->getUserData();
|
||||
|
||||
// Pack message
|
||||
auto msg = MsgReq{
|
||||
.connId = state->connId,
|
||||
.payload = std::move(json),
|
||||
};
|
||||
|
||||
// Dispatch to ReqWorker thread pool
|
||||
tpReqWorker->dispatchToThread(state->connId, std::move(msg));
|
||||
}
|
||||
```
|
||||
|
||||
**Message passing pattern:**
|
||||
|
||||
```cpp
|
||||
// ThreadPool::dispatchToThread
|
||||
void dispatchToThread(uint64_t connId, Message &&msg) {
|
||||
size_t threadId = connId % threads.size();
|
||||
threads[threadId]->queue.push(std::move(msg));
|
||||
}
|
||||
```
|
||||
|
||||
**Benefits:**
|
||||
- **Zero-copy:** `std::move` transfers ownership without copying
|
||||
- **Deterministic:** Same connection always processed by same thread
|
||||
- **Lock-free:** Each thread has own queue
|
||||
|
||||
## Event Ingestion Pipeline
|
||||
|
||||
### Ingester Thread Pool
|
||||
|
||||
```cpp
|
||||
void IngesterThread::run() {
|
||||
while (running) {
|
||||
Message msg;
|
||||
if (!queue.pop(msg, 100ms)) continue;
|
||||
|
||||
// Extract event from JSON
|
||||
auto event = parseEvent(msg.payload);
|
||||
|
||||
// Validate event ID
|
||||
if (!validateEventId(event)) {
|
||||
sendOK(msg.connId, event.id, false, "invalid: id mismatch");
|
||||
continue;
|
||||
}
|
||||
|
||||
// Verify signature (using thread-local secp256k1 context)
|
||||
if (!verifySignature(event, secpCtx)) {
|
||||
sendOK(msg.connId, event.id, false, "invalid: signature verification failed");
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check for duplicate (bloom filter + database)
|
||||
if (isDuplicate(event.id)) {
|
||||
sendOK(msg.connId, event.id, true, "duplicate: already have this event");
|
||||
continue;
|
||||
}
|
||||
|
||||
// Send to Writer thread
|
||||
auto writerMsg = MsgWriter{
|
||||
.connId = msg.connId,
|
||||
.event = std::move(event),
|
||||
};
|
||||
tpWriter->dispatch(std::move(writerMsg));
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Validation sequence:**
|
||||
1. Parse JSON into Event struct
|
||||
2. Validate event ID matches content hash
|
||||
3. Verify secp256k1 signature
|
||||
4. Check duplicate (bloom filter for speed)
|
||||
5. Forward to Writer thread for storage
|
||||
|
||||
### Writer Thread
|
||||
|
||||
```cpp
|
||||
void WriterThread::run() {
|
||||
// Single thread for all database writes
|
||||
while (running) {
|
||||
Message msg;
|
||||
if (!queue.pop(msg, 100ms)) continue;
|
||||
|
||||
// Write to database
|
||||
bool success = db.insertEvent(msg.event);
|
||||
|
||||
// Send OK to client
|
||||
sendOK(msg.connId, msg.event.id, success,
|
||||
success ? "" : "error: failed to store");
|
||||
|
||||
if (success) {
|
||||
// Broadcast to subscribers
|
||||
broadcastEvent(msg.event);
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Single-writer pattern:**
|
||||
- Only one thread writes to database
|
||||
- Eliminates write conflicts
|
||||
- Simplified transaction management
|
||||
|
||||
### Event Broadcasting
|
||||
|
||||
```cpp
|
||||
void broadcastEvent(const Event &event) {
|
||||
// Serialize event JSON once
|
||||
std::string eventJson = serializeEvent(event);
|
||||
|
||||
// Iterate all active subscriptions
|
||||
for (auto &[connId, sub] : activeSubscriptions) {
|
||||
// Check if filter matches
|
||||
if (!sub->filter.matches(event)) continue;
|
||||
|
||||
// Check if event newer than last sent
|
||||
if (event.id <= sub->latestEventSent) continue;
|
||||
|
||||
// Send to connection
|
||||
auto msg = MsgWebSocket{
|
||||
.connId = connId,
|
||||
.payload = eventJson, // Reuse serialized JSON
|
||||
};
|
||||
|
||||
tpWebSocket->dispatch(std::move(msg));
|
||||
|
||||
// Update latest sent
|
||||
sub->latestEventSent = event.id;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Critical optimization:** Serialize event JSON once, send to N subscribers
|
||||
|
||||
**Performance impact:** For 1000 subscribers, reduces:
|
||||
- JSON serialization: 1000× → 1×
|
||||
- Memory allocations: 1000× → 1×
|
||||
- CPU time: ~100ms → ~1ms
|
||||
|
||||
## Subscription Management
|
||||
|
||||
### REQ Processing
|
||||
|
||||
```cpp
|
||||
void ReqWorkerThread::run() {
|
||||
while (running) {
|
||||
MsgReq msg;
|
||||
if (!queue.pop(msg, 100ms)) continue;
|
||||
|
||||
// Parse REQ message: ["REQ", subId, filter1, filter2, ...]
|
||||
std::string subId = msg.payload[1];
|
||||
|
||||
// Create subscription object
|
||||
auto sub = std::make_shared<Subscription>();
|
||||
sub->subId = subId;
|
||||
|
||||
// Parse filters
|
||||
for (size_t i = 2; i < msg.payload.size(); i++) {
|
||||
Filter filter = parseFilter(msg.payload[i]);
|
||||
sub->filters.push_back(filter);
|
||||
}
|
||||
|
||||
// Store subscription
|
||||
activeSubscriptions[msg.connId] = sub;
|
||||
|
||||
// Query stored events
|
||||
std::vector<Event> events = db.queryEvents(sub->filters);
|
||||
|
||||
// Send matching events
|
||||
for (const auto &event : events) {
|
||||
sendEvent(msg.connId, subId, event);
|
||||
}
|
||||
|
||||
// Send EOSE
|
||||
sendEOSE(msg.connId, subId);
|
||||
|
||||
// Notify ReqMonitor to watch for real-time events
|
||||
auto monitorMsg = MsgReqMonitor{
|
||||
.connId = msg.connId,
|
||||
.subId = subId,
|
||||
};
|
||||
tpReqMonitor->dispatchToThread(msg.connId, std::move(monitorMsg));
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Query optimization:**
|
||||
|
||||
```cpp
|
||||
std::vector<Event> Database::queryEvents(const std::vector<Filter> &filters) {
|
||||
// Combine filters with OR logic
|
||||
std::string sql = "SELECT * FROM events WHERE ";
|
||||
|
||||
for (size_t i = 0; i < filters.size(); i++) {
|
||||
if (i > 0) sql += " OR ";
|
||||
sql += buildFilterSQL(filters[i]);
|
||||
}
|
||||
|
||||
sql += " ORDER BY created_at DESC LIMIT 1000";
|
||||
|
||||
return executeQuery(sql);
|
||||
}
|
||||
```
|
||||
|
||||
**Filter SQL generation:**
|
||||
|
||||
```cpp
|
||||
std::string buildFilterSQL(const Filter &filter) {
|
||||
std::vector<std::string> conditions;
|
||||
|
||||
// Event IDs
|
||||
if (!filter.ids.empty()) {
|
||||
conditions.push_back("id IN (" + joinQuoted(filter.ids) + ")");
|
||||
}
|
||||
|
||||
// Authors
|
||||
if (!filter.authors.empty()) {
|
||||
conditions.push_back("pubkey IN (" + joinQuoted(filter.authors) + ")");
|
||||
}
|
||||
|
||||
// Kinds
|
||||
if (!filter.kinds.empty()) {
|
||||
conditions.push_back("kind IN (" + join(filter.kinds) + ")");
|
||||
}
|
||||
|
||||
// Time range
|
||||
if (filter.since) {
|
||||
conditions.push_back("created_at >= " + std::to_string(*filter.since));
|
||||
}
|
||||
if (filter.until) {
|
||||
conditions.push_back("created_at <= " + std::to_string(*filter.until));
|
||||
}
|
||||
|
||||
// Tags (requires JOIN with tags table)
|
||||
if (!filter.tags.empty()) {
|
||||
for (const auto &[tagName, tagValues] : filter.tags) {
|
||||
conditions.push_back(
|
||||
"EXISTS (SELECT 1 FROM tags WHERE tags.event_id = events.id "
|
||||
"AND tags.name = '" + tagName + "' "
|
||||
"AND tags.value IN (" + joinQuoted(tagValues) + "))"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return "(" + join(conditions, " AND ") + ")";
|
||||
}
|
||||
```
|
||||
|
||||
### ReqMonitor for Real-Time Events
|
||||
|
||||
```cpp
|
||||
void ReqMonitorThread::run() {
|
||||
// Subscribe to event broadcast channel
|
||||
auto eventSubscription = subscribeToEvents();
|
||||
|
||||
while (running) {
|
||||
Event event;
|
||||
if (!eventSubscription.receive(event, 100ms)) continue;
|
||||
|
||||
// Check all subscriptions assigned to this thread
|
||||
for (auto &[connId, sub] : mySubscriptions) {
|
||||
// Only process subscriptions for this thread
|
||||
if (connId % numThreads != threadId) continue;
|
||||
|
||||
// Check if filter matches
|
||||
bool matches = false;
|
||||
for (const auto &filter : sub->filters) {
|
||||
if (filter.matches(event)) {
|
||||
matches = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (matches) {
|
||||
sendEvent(connId, sub->subId, event);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Pattern:** Monitor thread watches event stream, sends to matching subscriptions
|
||||
|
||||
### CLOSE Handling
|
||||
|
||||
```cpp
|
||||
void handleCloseMessage(auto *ws, nlohmann::json &&json) {
|
||||
auto *state = ws->getUserData();
|
||||
|
||||
// Parse CLOSE message: ["CLOSE", subId]
|
||||
std::string subId = json[1];
|
||||
|
||||
// Remove subscription
|
||||
activeSubscriptions.erase(state->connId);
|
||||
|
||||
LI << "Subscription closed: connId=" << state->connId
|
||||
<< " subId=" << subId;
|
||||
}
|
||||
```
|
||||
|
||||
## Performance Optimizations
|
||||
|
||||
### 1. Event Batching
|
||||
|
||||
**Problem:** Serializing same event 1000× for 1000 subscribers is wasteful
|
||||
|
||||
**Solution:** Serialize once, send to all
|
||||
|
||||
```cpp
|
||||
// BAD: Serialize for each subscriber
|
||||
for (auto &sub : subscriptions) {
|
||||
std::string json = serializeEvent(event); // Repeated!
|
||||
send(sub.connId, json);
|
||||
}
|
||||
|
||||
// GOOD: Serialize once
|
||||
std::string json = serializeEvent(event);
|
||||
for (auto &sub : subscriptions) {
|
||||
send(sub.connId, json); // Reuse!
|
||||
}
|
||||
```
|
||||
|
||||
**Measurement:** For 1000 subscribers, reduces broadcast time from 100ms to 1ms
|
||||
|
||||
### 2. Move Semantics
|
||||
|
||||
**Problem:** Copying large JSON objects is expensive
|
||||
|
||||
**Solution:** Transfer ownership with `std::move`
|
||||
|
||||
```cpp
|
||||
// BAD: Copies JSON object
|
||||
void dispatch(Message msg) {
|
||||
queue.push(msg); // Copy
|
||||
}
|
||||
|
||||
// GOOD: Moves JSON object
|
||||
void dispatch(Message &&msg) {
|
||||
queue.push(std::move(msg)); // Move
|
||||
}
|
||||
```
|
||||
|
||||
**Benefit:** Zero-copy message passing between threads
|
||||
|
||||
### 3. Pre-allocated Buffers
|
||||
|
||||
**Problem:** Allocating buffer for each message
|
||||
|
||||
**Solution:** Reuse buffer per connection
|
||||
|
||||
```cpp
|
||||
struct ConnectionState {
|
||||
std::string parseBuffer; // Reused for all messages
|
||||
};
|
||||
|
||||
void handleMessage(std::string_view msg) {
|
||||
state->parseBuffer.assign(msg.data(), msg.size());
|
||||
auto json = nlohmann::json::parse(state->parseBuffer);
|
||||
// ...
|
||||
}
|
||||
```
|
||||
|
||||
**Benefit:** Eliminates 10,000+ allocations/second per connection
|
||||
|
||||
### 4. std::variant for Message Types
|
||||
|
||||
**Problem:** Virtual function calls for polymorphic messages
|
||||
|
||||
**Solution:** `std::variant` with `std::visit`
|
||||
|
||||
```cpp
|
||||
// BAD: Virtual function (pointer indirection, vtable lookup)
|
||||
struct Message {
|
||||
virtual void handle() = 0;
|
||||
};
|
||||
|
||||
// GOOD: std::variant (no indirection, inlined)
|
||||
using Message = std::variant<
|
||||
MsgIngester,
|
||||
MsgReq,
|
||||
MsgWriter,
|
||||
MsgWebSocket
|
||||
>;
|
||||
|
||||
void handle(Message &&msg) {
|
||||
std::visit([](auto &&m) { m.handle(); }, msg);
|
||||
}
|
||||
```
|
||||
|
||||
**Benefit:** Compiler inlines visit, eliminates virtual call overhead
|
||||
|
||||
### 5. Bloom Filter for Duplicate Detection
|
||||
|
||||
**Problem:** Database query for every event to check duplicate
|
||||
|
||||
**Solution:** In-memory bloom filter for fast negative
|
||||
|
||||
```cpp
|
||||
class DuplicateDetector {
|
||||
BloomFilter bloom; // Fast probabilistic check
|
||||
|
||||
bool isDuplicate(const std::string &eventId) {
|
||||
// Fast negative (definitely not seen)
|
||||
if (!bloom.contains(eventId)) {
|
||||
bloom.insert(eventId);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Possible positive (maybe seen, check database)
|
||||
if (db.eventExists(eventId)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// False positive
|
||||
bloom.insert(eventId);
|
||||
return false;
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
**Benefit:** 99% of duplicate checks avoid database query
|
||||
|
||||
### 6. Batch Queue Operations
|
||||
|
||||
**Problem:** Lock contention on message queue
|
||||
|
||||
**Solution:** Batch multiple pushes with single lock
|
||||
|
||||
```cpp
|
||||
class MessageQueue {
|
||||
std::mutex mutex;
|
||||
std::deque<Message> queue;
|
||||
|
||||
void pushBatch(std::vector<Message> &messages) {
|
||||
std::lock_guard lock(mutex);
|
||||
for (auto &msg : messages) {
|
||||
queue.push_back(std::move(msg));
|
||||
}
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
**Benefit:** Reduces lock acquisitions by 10-100×
|
||||
|
||||
### 7. ZSTD Dictionary Compression
|
||||
|
||||
**Problem:** WebSocket compression slower than desired
|
||||
|
||||
**Solution:** Train ZSTD dictionary on typical Nostr messages
|
||||
|
||||
```cpp
|
||||
// Train dictionary on corpus of Nostr events
|
||||
std::string corpus = collectTypicalEvents();
|
||||
ZSTD_CDict *dict = ZSTD_createCDict(
|
||||
corpus.data(), corpus.size(),
|
||||
compressionLevel
|
||||
);
|
||||
|
||||
// Use dictionary for compression
|
||||
size_t compressedSize = ZSTD_compress_usingCDict(
|
||||
cctx, dst, dstSize,
|
||||
src, srcSize, dict
|
||||
);
|
||||
```
|
||||
|
||||
**Benefit:** 10-20% better compression ratio, 2× faster decompression
|
||||
|
||||
### 8. String Views
|
||||
|
||||
**Problem:** Unnecessary string copies when parsing
|
||||
|
||||
**Solution:** Use `std::string_view` for zero-copy
|
||||
|
||||
```cpp
|
||||
// BAD: Copies substring
|
||||
std::string extractCommand(const std::string &msg) {
|
||||
return msg.substr(0, 5); // Copy
|
||||
}
|
||||
|
||||
// GOOD: View into original string
|
||||
std::string_view extractCommand(std::string_view msg) {
|
||||
return msg.substr(0, 5); // No copy
|
||||
}
|
||||
```
|
||||
|
||||
**Benefit:** Eliminates allocations during parsing
|
||||
|
||||
## Compression (permessage-deflate)
|
||||
|
||||
### WebSocket Compression Configuration
|
||||
|
||||
```cpp
|
||||
struct PerMessageDeflate {
|
||||
z_stream deflate_stream;
|
||||
z_stream inflate_stream;
|
||||
|
||||
// Sliding window for compression history
|
||||
static constexpr int WINDOW_BITS = 15;
|
||||
static constexpr int MEM_LEVEL = 8;
|
||||
|
||||
void init() {
|
||||
// Initialize deflate (compression)
|
||||
deflate_stream.zalloc = Z_NULL;
|
||||
deflate_stream.zfree = Z_NULL;
|
||||
deflate_stream.opaque = Z_NULL;
|
||||
deflateInit2(&deflate_stream,
|
||||
Z_DEFAULT_COMPRESSION,
|
||||
Z_DEFLATED,
|
||||
-WINDOW_BITS, // Negative = no zlib header
|
||||
MEM_LEVEL,
|
||||
Z_DEFAULT_STRATEGY);
|
||||
|
||||
// Initialize inflate (decompression)
|
||||
inflate_stream.zalloc = Z_NULL;
|
||||
inflate_stream.zfree = Z_NULL;
|
||||
inflate_stream.opaque = Z_NULL;
|
||||
inflateInit2(&inflate_stream, -WINDOW_BITS);
|
||||
}
|
||||
|
||||
std::string compress(std::string_view data) {
|
||||
// Compress with sliding window
|
||||
deflate_stream.next_in = (Bytef*)data.data();
|
||||
deflate_stream.avail_in = data.size();
|
||||
|
||||
std::string compressed;
|
||||
compressed.resize(deflateBound(&deflate_stream, data.size()));
|
||||
|
||||
deflate_stream.next_out = (Bytef*)compressed.data();
|
||||
deflate_stream.avail_out = compressed.size();
|
||||
|
||||
deflate(&deflate_stream, Z_SYNC_FLUSH);
|
||||
|
||||
compressed.resize(compressed.size() - deflate_stream.avail_out);
|
||||
return compressed;
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
**Typical compression ratios:**
|
||||
- JSON events: 60-80% reduction
|
||||
- Subscription filters: 40-60% reduction
|
||||
- Binary events: 10-30% reduction
|
||||
|
||||
## Database Schema (LMDB)
|
||||
|
||||
strfry uses LMDB (Lightning Memory-Mapped Database) for event storage:
|
||||
|
||||
```cpp
|
||||
// Key-value stores
|
||||
struct EventDB {
|
||||
// Primary event storage (key: event ID, value: event data)
|
||||
lmdb::dbi eventsDB;
|
||||
|
||||
// Index by pubkey (key: pubkey + created_at, value: event ID)
|
||||
lmdb::dbi pubkeyDB;
|
||||
|
||||
// Index by kind (key: kind + created_at, value: event ID)
|
||||
lmdb::dbi kindDB;
|
||||
|
||||
// Index by tags (key: tag_name + tag_value + created_at, value: event ID)
|
||||
lmdb::dbi tagsDB;
|
||||
|
||||
// Deletion index (key: event ID, value: deletion event ID)
|
||||
lmdb::dbi deletionsDB;
|
||||
};
|
||||
```
|
||||
|
||||
**Why LMDB?**
|
||||
- Memory-mapped I/O (kernel manages caching)
|
||||
- Copy-on-write (MVCC without locks)
|
||||
- Ordered keys (enables range queries)
|
||||
- Crash-proof (no corruption on power loss)
|
||||
|
||||
## Monitoring and Metrics
|
||||
|
||||
### Connection Statistics
|
||||
|
||||
```cpp
|
||||
struct RelayStats {
|
||||
std::atomic<uint64_t> totalConnections{0};
|
||||
std::atomic<uint64_t> activeConnections{0};
|
||||
std::atomic<uint64_t> eventsReceived{0};
|
||||
std::atomic<uint64_t> eventsSent{0};
|
||||
std::atomic<uint64_t> bytesReceived{0};
|
||||
std::atomic<uint64_t> bytesSent{0};
|
||||
|
||||
void recordConnection() {
|
||||
totalConnections.fetch_add(1, std::memory_order_relaxed);
|
||||
activeConnections.fetch_add(1, std::memory_order_relaxed);
|
||||
}
|
||||
|
||||
void recordDisconnection() {
|
||||
activeConnections.fetch_sub(1, std::memory_order_relaxed);
|
||||
}
|
||||
|
||||
void recordEventReceived(size_t bytes) {
|
||||
eventsReceived.fetch_add(1, std::memory_order_relaxed);
|
||||
bytesReceived.fetch_add(bytes, std::memory_order_relaxed);
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
**Atomic operations:** Lock-free updates from multiple threads
|
||||
|
||||
### Performance Metrics
|
||||
|
||||
```cpp
|
||||
struct PerformanceMetrics {
|
||||
// Latency histograms
|
||||
Histogram eventIngestionLatency;
|
||||
Histogram subscriptionQueryLatency;
|
||||
Histogram eventBroadcastLatency;
|
||||
|
||||
// Thread pool queue depths
|
||||
std::atomic<size_t> ingesterQueueDepth{0};
|
||||
std::atomic<size_t> writerQueueDepth{0};
|
||||
std::atomic<size_t> reqWorkerQueueDepth{0};
|
||||
|
||||
void recordIngestion(std::chrono::microseconds duration) {
|
||||
eventIngestionLatency.record(duration.count());
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
### relay.conf Example
|
||||
|
||||
```ini
|
||||
[relay]
|
||||
bind = 0.0.0.0
|
||||
port = 8080
|
||||
maxConnections = 10000
|
||||
maxMessageSize = 16777216 # 16 MB
|
||||
|
||||
[ingester]
|
||||
threads = 3
|
||||
queueSize = 10000
|
||||
|
||||
[writer]
|
||||
threads = 1
|
||||
queueSize = 1000
|
||||
batchSize = 100
|
||||
|
||||
[reqWorker]
|
||||
threads = 3
|
||||
queueSize = 10000
|
||||
|
||||
[db]
|
||||
path = /var/lib/strfry/events.lmdb
|
||||
maxSizeGB = 100
|
||||
```
|
||||
|
||||
## Deployment Considerations
|
||||
|
||||
### System Limits
|
||||
|
||||
```bash
|
||||
# Increase file descriptor limit
|
||||
ulimit -n 65536
|
||||
|
||||
# Increase maximum socket connections
|
||||
sysctl -w net.core.somaxconn=4096
|
||||
|
||||
# TCP tuning
|
||||
sysctl -w net.ipv4.tcp_fin_timeout=15
|
||||
sysctl -w net.ipv4.tcp_tw_reuse=1
|
||||
```
|
||||
|
||||
### Memory Requirements
|
||||
|
||||
**Per connection:**
|
||||
- ConnectionState: ~1 KB
|
||||
- WebSocket buffers: ~32 KB (16 KB send + 16 KB receive)
|
||||
- Compression state: ~400 KB (200 KB deflate + 200 KB inflate)
|
||||
|
||||
**Total:** ~433 KB per connection
|
||||
|
||||
**For 10,000 connections:** ~4.3 GB
|
||||
|
||||
### CPU Requirements
|
||||
|
||||
**Single-core can handle:**
|
||||
- 1000 concurrent connections
|
||||
- 10,000 events/sec ingestion
|
||||
- 100,000 events/sec broadcast (cached)
|
||||
|
||||
**Recommended:**
|
||||
- 8+ cores for 10,000 connections
|
||||
- 16+ cores for 50,000 connections
|
||||
|
||||
## Summary
|
||||
|
||||
**Key architectural patterns:**
|
||||
1. **Single-threaded I/O:** epoll handles all connections in one thread
|
||||
2. **Specialized thread pools:** Different operations use dedicated threads
|
||||
3. **Deterministic assignment:** Connection ID determines thread assignment
|
||||
4. **Move semantics:** Zero-copy message passing
|
||||
5. **Event batching:** Serialize once, send to many
|
||||
6. **Pre-allocated buffers:** Reuse memory per connection
|
||||
7. **Bloom filters:** Fast duplicate detection
|
||||
8. **LMDB:** Memory-mapped database for zero-copy reads
|
||||
|
||||
**Performance characteristics:**
|
||||
- **50,000+ concurrent connections** per server
|
||||
- **100,000+ events/sec** throughput
|
||||
- **Sub-millisecond** latency for broadcasts
|
||||
- **10 GB+ event database** with fast queries
|
||||
|
||||
**When to use strfry patterns:**
|
||||
- Need maximum performance (trading complexity)
|
||||
- Have C++ expertise on team
|
||||
- Running large public relay (thousands of users)
|
||||
- Want minimal memory footprint
|
||||
- Need to scale to 50K+ connections
|
||||
|
||||
**Trade-offs:**
|
||||
- **Complexity:** More complex than Go/Rust implementations
|
||||
- **Portability:** Linux-specific (epoll, LMDB)
|
||||
- **Development speed:** Slower iteration than higher-level languages
|
||||
|
||||
**Further reading:**
|
||||
- strfry repository: https://github.com/hoytech/strfry
|
||||
- uWebSockets: https://github.com/uNetworking/uWebSockets
|
||||
- LMDB: http://www.lmdb.tech/doc/
|
||||
- epoll: https://man7.org/linux/man-pages/man7/epoll.7.html
|
||||
881
.claude/skills/nostr-websocket/references/websocket_protocol.md
Normal file
881
.claude/skills/nostr-websocket/references/websocket_protocol.md
Normal file
@@ -0,0 +1,881 @@
|
||||
# WebSocket Protocol (RFC 6455) - Complete Reference
|
||||
|
||||
## Connection Establishment
|
||||
|
||||
### HTTP Upgrade Handshake
|
||||
|
||||
The WebSocket protocol begins as an HTTP request that upgrades to WebSocket:
|
||||
|
||||
**Client Request:**
|
||||
```http
|
||||
GET /chat HTTP/1.1
|
||||
Host: server.example.com
|
||||
Upgrade: websocket
|
||||
Connection: Upgrade
|
||||
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
|
||||
Origin: http://example.com
|
||||
Sec-WebSocket-Protocol: chat, superchat
|
||||
Sec-WebSocket-Version: 13
|
||||
```
|
||||
|
||||
**Server Response:**
|
||||
```http
|
||||
HTTP/1.1 101 Switching Protocols
|
||||
Upgrade: websocket
|
||||
Connection: Upgrade
|
||||
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
|
||||
Sec-WebSocket-Protocol: chat
|
||||
```
|
||||
|
||||
### Handshake Details
|
||||
|
||||
**Sec-WebSocket-Key Generation (Client):**
|
||||
1. Generate 16 random bytes
|
||||
2. Base64-encode the result
|
||||
3. Send in `Sec-WebSocket-Key` header
|
||||
|
||||
**Sec-WebSocket-Accept Computation (Server):**
|
||||
1. Concatenate client key with GUID: `258EAFA5-E914-47DA-95CA-C5AB0DC85B11`
|
||||
2. Compute SHA-1 hash of concatenated string
|
||||
3. Base64-encode the hash
|
||||
4. Send in `Sec-WebSocket-Accept` header
|
||||
|
||||
**Example computation:**
|
||||
```
|
||||
Client Key: dGhlIHNhbXBsZSBub25jZQ==
|
||||
Concatenated: dGhlIHNhbXBsZSBub25jZQ==258EAFA5-E914-47DA-95CA-C5AB0DC85B11
|
||||
SHA-1 Hash: b37a4f2cc0cb4e7e8cf769a5f3f8f2e8e4c9f7a3
|
||||
Base64: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
|
||||
```
|
||||
|
||||
**Validation (Client):**
|
||||
- Verify HTTP status is 101
|
||||
- Verify `Sec-WebSocket-Accept` matches expected value
|
||||
- If validation fails, do not establish connection
|
||||
|
||||
### Origin Header
|
||||
|
||||
The `Origin` header provides protection against cross-site WebSocket hijacking:
|
||||
|
||||
**Server-side validation:**
|
||||
```go
|
||||
func checkOrigin(r *http.Request) bool {
|
||||
origin := r.Header.Get("Origin")
|
||||
allowedOrigins := []string{
|
||||
"https://example.com",
|
||||
"https://app.example.com",
|
||||
}
|
||||
for _, allowed := range allowedOrigins {
|
||||
if origin == allowed {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
```
|
||||
|
||||
**Security consideration:** Browser-based clients MUST send Origin header. Non-browser clients MAY omit it. Servers SHOULD validate Origin for browser clients to prevent CSRF attacks.
|
||||
|
||||
## Frame Format
|
||||
|
||||
### Base Framing Protocol
|
||||
|
||||
WebSocket frames use a binary format with variable-length fields:
|
||||
|
||||
```
|
||||
0 1 2 3
|
||||
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
||||
+-+-+-+-+-------+-+-------------+-------------------------------+
|
||||
|F|R|R|R| opcode|M| Payload len | Extended payload length |
|
||||
|I|S|S|S| (4) |A| (7) | (16/64) |
|
||||
|N|V|V|V| |S| | (if payload len==126/127) |
|
||||
| |1|2|3| |K| | |
|
||||
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
|
||||
| Extended payload length continued, if payload len == 127 |
|
||||
+ - - - - - - - - - - - - - - - +-------------------------------+
|
||||
| |Masking-key, if MASK set to 1 |
|
||||
+-------------------------------+-------------------------------+
|
||||
| Masking-key (continued) | Payload Data |
|
||||
+-------------------------------- - - - - - - - - - - - - - - - +
|
||||
: Payload Data continued ... :
|
||||
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|
||||
| Payload Data continued ... |
|
||||
+---------------------------------------------------------------+
|
||||
```
|
||||
|
||||
### Frame Header Fields
|
||||
|
||||
**FIN (1 bit):**
|
||||
- `1` = Final fragment in message
|
||||
- `0` = More fragments follow
|
||||
- Used for message fragmentation
|
||||
|
||||
**RSV1, RSV2, RSV3 (1 bit each):**
|
||||
- Reserved for extensions
|
||||
- MUST be 0 unless extension negotiated
|
||||
- Server MUST fail connection if non-zero with no extension
|
||||
|
||||
**Opcode (4 bits):**
|
||||
- Defines interpretation of payload data
|
||||
- See "Frame Opcodes" section below
|
||||
|
||||
**MASK (1 bit):**
|
||||
- `1` = Payload is masked (required for client-to-server)
|
||||
- `0` = Payload is not masked (required for server-to-client)
|
||||
- Client MUST mask all frames sent to server
|
||||
- Server MUST NOT mask frames sent to client
|
||||
|
||||
**Payload Length (7 bits, 7+16 bits, or 7+64 bits):**
|
||||
- If 0-125: Actual payload length
|
||||
- If 126: Next 2 bytes are 16-bit unsigned payload length
|
||||
- If 127: Next 8 bytes are 64-bit unsigned payload length
|
||||
|
||||
**Masking-key (0 or 4 bytes):**
|
||||
- Present if MASK bit is set
|
||||
- 32-bit value used to mask payload
|
||||
- MUST be unpredictable (strong entropy source)
|
||||
|
||||
### Frame Opcodes
|
||||
|
||||
**Data Frame Opcodes:**
|
||||
- `0x0` - Continuation Frame
|
||||
- Used for fragmented messages
|
||||
- Must follow initial data frame (text/binary)
|
||||
- Carries same data type as initial frame
|
||||
|
||||
- `0x1` - Text Frame
|
||||
- Payload is UTF-8 encoded text
|
||||
- MUST be valid UTF-8
|
||||
- Endpoint MUST fail connection if invalid UTF-8
|
||||
|
||||
- `0x2` - Binary Frame
|
||||
- Payload is arbitrary binary data
|
||||
- Application interprets data
|
||||
|
||||
- `0x3-0x7` - Reserved for future non-control frames
|
||||
|
||||
**Control Frame Opcodes:**
|
||||
- `0x8` - Connection Close
|
||||
- Initiates or acknowledges connection closure
|
||||
- MAY contain status code and reason
|
||||
- See "Close Handshake" section
|
||||
|
||||
- `0x9` - Ping
|
||||
- Heartbeat mechanism
|
||||
- MAY contain application data
|
||||
- Recipient MUST respond with Pong
|
||||
|
||||
- `0xA` - Pong
|
||||
- Response to Ping
|
||||
- MUST contain identical payload as Ping
|
||||
- MAY be sent unsolicited (unidirectional heartbeat)
|
||||
|
||||
- `0xB-0xF` - Reserved for future control frames
|
||||
|
||||
### Control Frame Constraints
|
||||
|
||||
**Control frames are subject to strict rules:**
|
||||
|
||||
1. **Maximum payload:** 125 bytes
|
||||
- Allows control frames to fit in single IP packet
|
||||
- Reduces fragmentation
|
||||
|
||||
2. **No fragmentation:** Control frames MUST NOT be fragmented
|
||||
- FIN bit MUST be 1
|
||||
- Ensures immediate processing
|
||||
|
||||
3. **Interleaving:** Control frames MAY be injected in middle of fragmented message
|
||||
- Enables ping/pong during long transfers
|
||||
- Close frames can interrupt any operation
|
||||
|
||||
4. **All control frames MUST be handled immediately**
|
||||
|
||||
### Masking
|
||||
|
||||
**Purpose of masking:**
|
||||
- Prevents cache poisoning attacks
|
||||
- Protects against misinterpretation by intermediaries
|
||||
- Makes WebSocket traffic unpredictable to proxies
|
||||
|
||||
**Masking algorithm:**
|
||||
```
|
||||
j = i MOD 4
|
||||
transformed-octet-i = original-octet-i XOR masking-key-octet-j
|
||||
```
|
||||
|
||||
**Implementation:**
|
||||
```go
|
||||
func maskBytes(data []byte, mask [4]byte) {
|
||||
for i := range data {
|
||||
data[i] ^= mask[i%4]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Example:**
|
||||
```
|
||||
Original: [0x48, 0x65, 0x6C, 0x6C, 0x6F] // "Hello"
|
||||
Masking Key: [0x37, 0xFA, 0x21, 0x3D]
|
||||
Masked: [0x7F, 0x9F, 0x4D, 0x51, 0x58]
|
||||
|
||||
Calculation:
|
||||
0x48 XOR 0x37 = 0x7F
|
||||
0x65 XOR 0xFA = 0x9F
|
||||
0x6C XOR 0x21 = 0x4D
|
||||
0x6C XOR 0x3D = 0x51
|
||||
0x6F XOR 0x37 = 0x58 (wraps around to mask[0])
|
||||
```
|
||||
|
||||
**Security requirement:** Masking key MUST be derived from strong source of entropy. Predictable masking keys defeat the security purpose.
|
||||
|
||||
## Message Fragmentation
|
||||
|
||||
### Why Fragment?
|
||||
|
||||
- Send message without knowing total size upfront
|
||||
- Multiplex logical channels (interleave messages)
|
||||
- Keep control frames responsive during large transfers
|
||||
|
||||
### Fragmentation Rules
|
||||
|
||||
**Sender rules:**
|
||||
1. First fragment has opcode (text/binary)
|
||||
2. Subsequent fragments have opcode 0x0 (continuation)
|
||||
3. Last fragment has FIN bit set to 1
|
||||
4. Control frames MAY be interleaved
|
||||
|
||||
**Receiver rules:**
|
||||
1. Reassemble fragments in order
|
||||
2. Final message type determined by first fragment opcode
|
||||
3. Validate UTF-8 across all text fragments
|
||||
4. Process control frames immediately (don't wait for FIN)
|
||||
|
||||
### Fragmentation Example
|
||||
|
||||
**Sending "Hello World" in 3 fragments:**
|
||||
|
||||
```
|
||||
Frame 1 (Text, More Fragments):
|
||||
FIN=0, Opcode=0x1, Payload="Hello"
|
||||
|
||||
Frame 2 (Continuation, More Fragments):
|
||||
FIN=0, Opcode=0x0, Payload=" Wor"
|
||||
|
||||
Frame 3 (Continuation, Final):
|
||||
FIN=1, Opcode=0x0, Payload="ld"
|
||||
```
|
||||
|
||||
**With interleaved Ping:**
|
||||
|
||||
```
|
||||
Frame 1: FIN=0, Opcode=0x1, Payload="Hello"
|
||||
Frame 2: FIN=1, Opcode=0x9, Payload="" <- Ping (complete)
|
||||
Frame 3: FIN=0, Opcode=0x0, Payload=" Wor"
|
||||
Frame 4: FIN=1, Opcode=0x0, Payload="ld"
|
||||
```
|
||||
|
||||
### Implementation Pattern
|
||||
|
||||
```go
|
||||
type fragmentState struct {
|
||||
messageType int
|
||||
fragments [][]byte
|
||||
}
|
||||
|
||||
func (ws *WebSocket) handleFrame(fin bool, opcode int, payload []byte) {
|
||||
switch opcode {
|
||||
case 0x1, 0x2: // Text or Binary (first fragment)
|
||||
if fin {
|
||||
ws.handleCompleteMessage(opcode, payload)
|
||||
} else {
|
||||
ws.fragmentState = &fragmentState{
|
||||
messageType: opcode,
|
||||
fragments: [][]byte{payload},
|
||||
}
|
||||
}
|
||||
|
||||
case 0x0: // Continuation
|
||||
if ws.fragmentState == nil {
|
||||
ws.fail("Unexpected continuation frame")
|
||||
return
|
||||
}
|
||||
ws.fragmentState.fragments = append(ws.fragmentState.fragments, payload)
|
||||
if fin {
|
||||
complete := bytes.Join(ws.fragmentState.fragments, nil)
|
||||
ws.handleCompleteMessage(ws.fragmentState.messageType, complete)
|
||||
ws.fragmentState = nil
|
||||
}
|
||||
|
||||
case 0x8, 0x9, 0xA: // Control frames
|
||||
ws.handleControlFrame(opcode, payload)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Ping and Pong Frames
|
||||
|
||||
### Purpose
|
||||
|
||||
1. **Keep-alive:** Detect broken connections
|
||||
2. **Latency measurement:** Time round-trip
|
||||
3. **NAT traversal:** Maintain mapping in stateful firewalls
|
||||
|
||||
### Protocol Rules
|
||||
|
||||
**Ping (0x9):**
|
||||
- MAY be sent by either endpoint at any time
|
||||
- MAY contain application data (≤125 bytes)
|
||||
- Application data arbitrary (often empty or timestamp)
|
||||
|
||||
**Pong (0xA):**
|
||||
- MUST be sent in response to Ping
|
||||
- MUST contain identical payload as Ping
|
||||
- MUST be sent "as soon as practical"
|
||||
- MAY be sent unsolicited (one-way heartbeat)
|
||||
|
||||
**No Response:**
|
||||
- If Pong not received within timeout, connection assumed dead
|
||||
- Application should close connection
|
||||
|
||||
### Implementation Patterns
|
||||
|
||||
**Pattern 1: Automatic Pong (most WebSocket libraries)**
|
||||
```go
|
||||
// Library handles pong automatically
|
||||
ws.SetPingHandler(func(appData string) error {
|
||||
// Custom handler if needed
|
||||
return nil // Library sends pong automatically
|
||||
})
|
||||
```
|
||||
|
||||
**Pattern 2: Manual Pong**
|
||||
```go
|
||||
func (ws *WebSocket) handlePing(payload []byte) {
|
||||
pongFrame := Frame{
|
||||
FIN: true,
|
||||
Opcode: 0xA,
|
||||
Payload: payload, // Echo same payload
|
||||
}
|
||||
ws.writeFrame(pongFrame)
|
||||
}
|
||||
```
|
||||
|
||||
**Pattern 3: Periodic Client Ping**
|
||||
```go
|
||||
func (ws *WebSocket) pingLoop() {
|
||||
ticker := time.NewTicker(30 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if err := ws.writePing([]byte{}); err != nil {
|
||||
return // Connection dead
|
||||
}
|
||||
case <-ws.done:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Pattern 4: Timeout Detection**
|
||||
```go
|
||||
const pongWait = 60 * time.Second
|
||||
|
||||
ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||
ws.SetPongHandler(func(string) error {
|
||||
ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||
return nil
|
||||
})
|
||||
|
||||
// If no frame received in pongWait, ReadMessage returns timeout error
|
||||
```
|
||||
|
||||
### Nostr Relay Recommendations
|
||||
|
||||
**Server-side:**
|
||||
- Send ping every 30-60 seconds
|
||||
- Close connection if no pong within 60-120 seconds
|
||||
- Log timeout closures for monitoring
|
||||
|
||||
**Client-side:**
|
||||
- Respond to pings automatically (use library handler)
|
||||
- Consider sending unsolicited pongs every 30 seconds (some proxies)
|
||||
- Reconnect if no frames received for 120 seconds
|
||||
|
||||
## Close Handshake
|
||||
|
||||
### Close Frame Structure
|
||||
|
||||
**Close frame (Opcode 0x8) payload:**
|
||||
```
|
||||
0 1 2 3
|
||||
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
||||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||||
| Status Code (16) | Reason (variable length)... |
|
||||
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
||||
```
|
||||
|
||||
**Status Code (2 bytes, optional):**
|
||||
- 16-bit unsigned integer
|
||||
- Network byte order (big-endian)
|
||||
- See "Status Codes" section below
|
||||
|
||||
**Reason (variable length, optional):**
|
||||
- UTF-8 encoded text
|
||||
- MUST be valid UTF-8
|
||||
- Typically human-readable explanation
|
||||
|
||||
### Close Handshake Sequence
|
||||
|
||||
**Initiator (either endpoint):**
|
||||
1. Send Close frame with optional status/reason
|
||||
2. Stop sending data frames
|
||||
3. Continue processing received frames until Close frame received
|
||||
4. Close underlying TCP connection
|
||||
|
||||
**Recipient:**
|
||||
1. Receive Close frame
|
||||
2. Send Close frame in response (if not already sent)
|
||||
3. Close underlying TCP connection
|
||||
|
||||
### Status Codes
|
||||
|
||||
**Normal Closure Codes:**
|
||||
- `1000` - Normal Closure
|
||||
- Successful operation complete
|
||||
- Default if no code specified
|
||||
|
||||
- `1001` - Going Away
|
||||
- Endpoint going away (server shutdown, browser navigation)
|
||||
- Client navigating to new page
|
||||
|
||||
**Error Closure Codes:**
|
||||
- `1002` - Protocol Error
|
||||
- Endpoint terminating due to protocol error
|
||||
- Invalid frame format, unexpected opcode, etc.
|
||||
|
||||
- `1003` - Unsupported Data
|
||||
- Endpoint cannot accept data type
|
||||
- Server received binary when expecting text
|
||||
|
||||
- `1007` - Invalid Frame Payload Data
|
||||
- Inconsistent data (e.g., non-UTF-8 in text frame)
|
||||
|
||||
- `1008` - Policy Violation
|
||||
- Message violates endpoint policy
|
||||
- Generic code when specific code doesn't fit
|
||||
|
||||
- `1009` - Message Too Big
|
||||
- Message too large to process
|
||||
|
||||
- `1010` - Mandatory Extension
|
||||
- Client expected server to negotiate extension
|
||||
- Server didn't respond with extension
|
||||
|
||||
- `1011` - Internal Server Error
|
||||
- Server encountered unexpected condition
|
||||
- Prevents fulfilling request
|
||||
|
||||
**Reserved Codes:**
|
||||
- `1004` - Reserved
|
||||
- `1005` - No Status Rcvd (internal use only, never sent)
|
||||
- `1006` - Abnormal Closure (internal use only, never sent)
|
||||
- `1015` - TLS Handshake (internal use only, never sent)
|
||||
|
||||
**Custom Application Codes:**
|
||||
- `3000-3999` - Library/framework use
|
||||
- `4000-4999` - Application use (e.g., Nostr-specific)
|
||||
|
||||
### Implementation Patterns
|
||||
|
||||
**Graceful close (initiator):**
|
||||
```go
|
||||
func (ws *WebSocket) Close() error {
|
||||
// Send close frame
|
||||
closeFrame := Frame{
|
||||
FIN: true,
|
||||
Opcode: 0x8,
|
||||
Payload: encodeCloseStatus(1000, "goodbye"),
|
||||
}
|
||||
ws.writeFrame(closeFrame)
|
||||
|
||||
// Wait for close frame response (with timeout)
|
||||
ws.SetReadDeadline(time.Now().Add(5 * time.Second))
|
||||
for {
|
||||
frame, err := ws.readFrame()
|
||||
if err != nil || frame.Opcode == 0x8 {
|
||||
break
|
||||
}
|
||||
// Process other frames
|
||||
}
|
||||
|
||||
// Close TCP connection
|
||||
return ws.conn.Close()
|
||||
}
|
||||
```
|
||||
|
||||
**Handling received close:**
|
||||
```go
|
||||
func (ws *WebSocket) handleCloseFrame(payload []byte) {
|
||||
status, reason := decodeClosePayload(payload)
|
||||
log.Printf("Close received: %d %s", status, reason)
|
||||
|
||||
// Send close response
|
||||
closeFrame := Frame{
|
||||
FIN: true,
|
||||
Opcode: 0x8,
|
||||
Payload: payload, // Echo same status/reason
|
||||
}
|
||||
ws.writeFrame(closeFrame)
|
||||
|
||||
// Close connection
|
||||
ws.conn.Close()
|
||||
}
|
||||
```
|
||||
|
||||
**Nostr relay close examples:**
|
||||
```go
|
||||
// Client subscription limit exceeded
|
||||
ws.SendClose(4000, "subscription limit exceeded")
|
||||
|
||||
// Invalid message format
|
||||
ws.SendClose(1002, "protocol error: invalid JSON")
|
||||
|
||||
// Relay shutting down
|
||||
ws.SendClose(1001, "relay shutting down")
|
||||
|
||||
// Client rate limit exceeded
|
||||
ws.SendClose(4001, "rate limit exceeded")
|
||||
```
|
||||
|
||||
## Security Considerations
|
||||
|
||||
### Origin-Based Security Model
|
||||
|
||||
**Threat:** Malicious web page opens WebSocket to victim server using user's credentials
|
||||
|
||||
**Mitigation:**
|
||||
1. Server checks `Origin` header
|
||||
2. Reject connections from untrusted origins
|
||||
3. Implement same-origin or allowlist policy
|
||||
|
||||
**Example:**
|
||||
```go
|
||||
func validateOrigin(r *http.Request) bool {
|
||||
origin := r.Header.Get("Origin")
|
||||
|
||||
// Allow same-origin
|
||||
if origin == "https://"+r.Host {
|
||||
return true
|
||||
}
|
||||
|
||||
// Allowlist trusted origins
|
||||
trusted := []string{
|
||||
"https://app.example.com",
|
||||
"https://mobile.example.com",
|
||||
}
|
||||
for _, t := range trusted {
|
||||
if origin == t {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
```
|
||||
|
||||
### Masking Attacks
|
||||
|
||||
**Why masking is required:**
|
||||
- Without masking, attacker can craft WebSocket frames that look like HTTP requests
|
||||
- Proxies might misinterpret frame data as HTTP
|
||||
- Could lead to cache poisoning or request smuggling
|
||||
|
||||
**Example attack (without masking):**
|
||||
```
|
||||
WebSocket payload: "GET /admin HTTP/1.1\r\nHost: victim.com\r\n\r\n"
|
||||
Proxy might interpret as separate HTTP request
|
||||
```
|
||||
|
||||
**Defense:** Client MUST mask all frames. Server MUST reject unmasked frames from client.
|
||||
|
||||
### Connection Limits
|
||||
|
||||
**Prevent resource exhaustion:**
|
||||
|
||||
```go
|
||||
type ConnectionLimiter struct {
|
||||
connections map[string]int
|
||||
maxPerIP int
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func (cl *ConnectionLimiter) Allow(ip string) bool {
|
||||
cl.mu.Lock()
|
||||
defer cl.mu.Unlock()
|
||||
|
||||
if cl.connections[ip] >= cl.maxPerIP {
|
||||
return false
|
||||
}
|
||||
cl.connections[ip]++
|
||||
return true
|
||||
}
|
||||
|
||||
func (cl *ConnectionLimiter) Release(ip string) {
|
||||
cl.mu.Lock()
|
||||
defer cl.mu.Unlock()
|
||||
cl.connections[ip]--
|
||||
}
|
||||
```
|
||||
|
||||
### TLS (WSS)
|
||||
|
||||
**Use WSS (WebSocket Secure) for:**
|
||||
- Authentication credentials
|
||||
- Private user data
|
||||
- Financial transactions
|
||||
- Any sensitive information
|
||||
|
||||
**WSS connection flow:**
|
||||
1. Establish TLS connection
|
||||
2. Perform TLS handshake
|
||||
3. Verify server certificate
|
||||
4. Perform WebSocket handshake over TLS
|
||||
|
||||
**URL schemes:**
|
||||
- `ws://` - Unencrypted WebSocket (default port 80)
|
||||
- `wss://` - Encrypted WebSocket over TLS (default port 443)
|
||||
|
||||
### Message Size Limits
|
||||
|
||||
**Prevent memory exhaustion:**
|
||||
|
||||
```go
|
||||
const maxMessageSize = 512 * 1024 // 512 KB
|
||||
|
||||
ws.SetReadLimit(maxMessageSize)
|
||||
|
||||
// Or during frame reading:
|
||||
if payloadLength > maxMessageSize {
|
||||
ws.SendClose(1009, "message too large")
|
||||
ws.Close()
|
||||
}
|
||||
```
|
||||
|
||||
### Rate Limiting
|
||||
|
||||
**Prevent abuse:**
|
||||
|
||||
```go
|
||||
type RateLimiter struct {
|
||||
limiter *rate.Limiter
|
||||
}
|
||||
|
||||
func (rl *RateLimiter) Allow() bool {
|
||||
return rl.limiter.Allow()
|
||||
}
|
||||
|
||||
// Per-connection limiter
|
||||
limiter := rate.NewLimiter(10, 20) // 10 msgs/sec, burst 20
|
||||
|
||||
if !limiter.Allow() {
|
||||
ws.SendClose(4001, "rate limit exceeded")
|
||||
}
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
### Connection Errors
|
||||
|
||||
**Types of errors:**
|
||||
1. **Network errors:** TCP connection failure, timeout
|
||||
2. **Protocol errors:** Invalid frame format, wrong opcode
|
||||
3. **Application errors:** Invalid message content
|
||||
|
||||
**Handling strategy:**
|
||||
```go
|
||||
for {
|
||||
frame, err := ws.ReadFrame()
|
||||
if err != nil {
|
||||
// Check error type
|
||||
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
|
||||
// Timeout - connection likely dead
|
||||
log.Println("Connection timeout")
|
||||
ws.Close()
|
||||
return
|
||||
}
|
||||
|
||||
if err == io.EOF || err == io.ErrUnexpectedEOF {
|
||||
// Connection closed
|
||||
log.Println("Connection closed")
|
||||
return
|
||||
}
|
||||
|
||||
if protocolErr, ok := err.(*ProtocolError); ok {
|
||||
// Protocol violation
|
||||
log.Printf("Protocol error: %v", protocolErr)
|
||||
ws.SendClose(1002, protocolErr.Error())
|
||||
ws.Close()
|
||||
return
|
||||
}
|
||||
|
||||
// Unknown error
|
||||
log.Printf("Unknown error: %v", err)
|
||||
ws.Close()
|
||||
return
|
||||
}
|
||||
|
||||
// Process frame
|
||||
}
|
||||
```
|
||||
|
||||
### UTF-8 Validation
|
||||
|
||||
**Text frames MUST contain valid UTF-8:**
|
||||
|
||||
```go
|
||||
func validateUTF8(data []byte) bool {
|
||||
return utf8.Valid(data)
|
||||
}
|
||||
|
||||
func handleTextFrame(payload []byte) error {
|
||||
if !validateUTF8(payload) {
|
||||
return fmt.Errorf("invalid UTF-8 in text frame")
|
||||
}
|
||||
// Process valid text
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
**For fragmented messages:** Validate UTF-8 across all fragments when reassembled.
|
||||
|
||||
## Implementation Checklist
|
||||
|
||||
### Client Implementation
|
||||
|
||||
- [ ] Generate random Sec-WebSocket-Key
|
||||
- [ ] Compute and validate Sec-WebSocket-Accept
|
||||
- [ ] MUST mask all frames sent to server
|
||||
- [ ] Handle unmasked frames from server
|
||||
- [ ] Respond to Ping with Pong
|
||||
- [ ] Implement close handshake (both initiating and responding)
|
||||
- [ ] Validate UTF-8 in text frames
|
||||
- [ ] Handle fragmented messages
|
||||
- [ ] Set reasonable timeouts
|
||||
- [ ] Implement reconnection logic
|
||||
|
||||
### Server Implementation
|
||||
|
||||
- [ ] Validate Sec-WebSocket-Key format
|
||||
- [ ] Compute correct Sec-WebSocket-Accept
|
||||
- [ ] Validate Origin header
|
||||
- [ ] MUST NOT mask frames sent to client
|
||||
- [ ] Reject masked frames from server (protocol error)
|
||||
- [ ] Respond to Ping with Pong
|
||||
- [ ] Implement close handshake (both initiating and responding)
|
||||
- [ ] Validate UTF-8 in text frames
|
||||
- [ ] Handle fragmented messages
|
||||
- [ ] Implement connection limits (per IP, total)
|
||||
- [ ] Implement message size limits
|
||||
- [ ] Implement rate limiting
|
||||
- [ ] Log connection statistics
|
||||
- [ ] Graceful shutdown (close all connections)
|
||||
|
||||
### Both Client and Server
|
||||
|
||||
- [ ] Handle concurrent read/write safely
|
||||
- [ ] Process control frames immediately (even during fragmentation)
|
||||
- [ ] Implement proper timeout mechanisms
|
||||
- [ ] Log errors with appropriate detail
|
||||
- [ ] Handle unexpected close gracefully
|
||||
- [ ] Validate frame structure
|
||||
- [ ] Check RSV bits (must be 0 unless extension)
|
||||
- [ ] Support standard close status codes
|
||||
- [ ] Implement proper error handling for all operations
|
||||
|
||||
## Common Implementation Mistakes
|
||||
|
||||
### 1. Concurrent Writes
|
||||
|
||||
**Mistake:** Writing to WebSocket from multiple goroutines without synchronization
|
||||
|
||||
**Fix:** Use mutex or single-writer goroutine
|
||||
```go
|
||||
type WebSocket struct {
|
||||
conn *websocket.Conn
|
||||
mutex sync.Mutex
|
||||
}
|
||||
|
||||
func (ws *WebSocket) WriteMessage(data []byte) error {
|
||||
ws.mutex.Lock()
|
||||
defer ws.mutex.Unlock()
|
||||
return ws.conn.WriteMessage(websocket.TextMessage, data)
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Not Handling Pong
|
||||
|
||||
**Mistake:** Sending Ping but not updating read deadline on Pong
|
||||
|
||||
**Fix:**
|
||||
```go
|
||||
ws.SetPongHandler(func(string) error {
|
||||
ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||
return nil
|
||||
})
|
||||
```
|
||||
|
||||
### 3. Forgetting Close Handshake
|
||||
|
||||
**Mistake:** Just calling `conn.Close()` without sending Close frame
|
||||
|
||||
**Fix:** Send Close frame first, wait for response, then close TCP
|
||||
|
||||
### 4. Not Validating UTF-8
|
||||
|
||||
**Mistake:** Accepting any bytes in text frames
|
||||
|
||||
**Fix:** Validate UTF-8 and fail connection on invalid text
|
||||
|
||||
### 5. No Message Size Limit
|
||||
|
||||
**Mistake:** Allowing unlimited message sizes
|
||||
|
||||
**Fix:** Set `SetReadLimit()` to reasonable value (e.g., 512 KB)
|
||||
|
||||
### 6. Blocking on Write
|
||||
|
||||
**Mistake:** Blocking indefinitely on slow clients
|
||||
|
||||
**Fix:** Set write deadline before each write
|
||||
```go
|
||||
ws.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
||||
```
|
||||
|
||||
### 7. Memory Leaks
|
||||
|
||||
**Mistake:** Not cleaning up resources on disconnect
|
||||
|
||||
**Fix:** Use defer for cleanup, ensure all goroutines terminate
|
||||
|
||||
### 8. Race Conditions in Close
|
||||
|
||||
**Mistake:** Multiple goroutines trying to close connection
|
||||
|
||||
**Fix:** Use `sync.Once` for close operation
|
||||
```go
|
||||
type WebSocket struct {
|
||||
conn *websocket.Conn
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
func (ws *WebSocket) Close() error {
|
||||
var err error
|
||||
ws.closeOnce.Do(func() {
|
||||
err = ws.conn.Close()
|
||||
})
|
||||
return err
|
||||
}
|
||||
```
|
||||
Reference in New Issue
Block a user