From f2ea19a8d4c8ea5f5e7e567131d5e09e9314acbc Mon Sep 17 00:00:00 2001 From: Strange Signal Date: Sat, 28 Feb 2026 18:29:22 -0500 Subject: [PATCH 1/3] feat: add optional zstd event payload compression at ingest --- golpe.yaml | 1 + src/events.cpp | 59 ++++++++++++++++++++++++++++++++++++++++++++++++-- strfry.conf | 5 +++++ 3 files changed, 63 insertions(+), 2 deletions(-) diff --git a/golpe.yaml b/golpe.yaml index 37f97276..2f79c212 100644 --- a/golpe.yaml +++ b/golpe.yaml @@ -89,6 +89,7 @@ tablesRaw: ## vals are prefixed with a type byte: ## 0: no compression, payload follows ## 1: zstd compression. Followed by Dictionary ID (native endian uint32) then compressed payload + ## 2: zstd compression (no dictionary), compressed payload follows EventPayload: flags: 'MDB_INTEGERKEY' diff --git a/src/events.cpp b/src/events.cpp index aadb6186..45a9599a 100644 --- a/src/events.cpp +++ b/src/events.cpp @@ -1,10 +1,29 @@ #include #include +#include +#include #include "events.h" #include "jsonParseUtils.h" +static int getPayloadZstdLevel() { + static int level = []() { + const char *s = std::getenv("STRFRY_EVENT_PAYLOAD_ZSTD_LEVEL"); + if (!s) return 0; + + try { + int v = std::stoi(s); + return std::max(0, std::min(v, 22)); + } catch (...) { + return 0; + } + }(); + + return level; +} + + std::string nostrJsonToPackedEvent(const tao::json::value &v) { PackedEventTagBuilder tagBuilder; @@ -212,6 +231,23 @@ std::string_view decodeEventPayload(lmdb::txn &txn, Decompressor &decomp, std::s if (outDictId) *outDictId = dictId; if (outCompressedSize) *outCompressedSize = raw.size(); return buf; + } else if (raw[0] == '\x02') { + raw = raw.substr(1); + + size_t frameSize = ZSTD_getFrameContentSize(raw.data(), raw.size()); + if (frameSize == ZSTD_CONTENTSIZE_ERROR) throw herr("EventPayload zstd frame invalid"); + + size_t outSize = cfg().events__maxEventSize; + if (frameSize != ZSTD_CONTENTSIZE_UNKNOWN) outSize = std::min(frameSize, cfg().events__maxEventSize); + + decomp.reserve(outSize); + + auto ret = ZSTD_decompress(decomp.buffer.data(), decomp.buffer.size(), raw.data(), raw.size()); + if (ZSTD_isError(ret)) throw herr("zstd decompression failed: ", ZSTD_getErrorName(ret)); + + if (outDictId) *outDictId = 0; + if (outCompressedSize) *outCompressedSize = raw.size(); + return std::string_view(decomp.buffer.data(), ret); } else { throw herr("Unexpected first byte in EventPayload"); } @@ -328,8 +364,27 @@ void writeEvents(lmdb::txn &txn, NegentropyFilterCache &neFilterCache, std::vect ev.levId = env.insert_Event(txn, ev.packedStr); tmpBuf.clear(); - tmpBuf += '\x00'; - tmpBuf += ev.jsonStr; + + int zstdLevel = getPayloadZstdLevel(); + if (zstdLevel > 0 && ev.jsonStr.size() > 64) { + std::string compressed; + compressed.resize(ZSTD_compressBound(ev.jsonStr.size())); + + auto ret = ZSTD_compress(compressed.data(), compressed.size(), ev.jsonStr.data(), ev.jsonStr.size(), zstdLevel); + + if (!ZSTD_isError(ret) && (ret + 1) < ev.jsonStr.size()) { + compressed.resize(ret); + tmpBuf += '\x02'; + tmpBuf += compressed; + } else { + tmpBuf += '\x00'; + tmpBuf += ev.jsonStr; + } + } else { + tmpBuf += '\x00'; + tmpBuf += ev.jsonStr; + } + env.dbi_EventPayload.put(txn, lmdb::to_sv(ev.levId), tmpBuf); updateNegentropy(PackedEventView(ev.packedStr), true); diff --git a/strfry.conf b/strfry.conf index 8824903a..3ce2c743 100644 --- a/strfry.conf +++ b/strfry.conf @@ -37,6 +37,11 @@ events { # Maximum size for tag values, in bytes maxTagValSize = 1024 + + # Optional event-payload zstd compression at ingest (experimental). + # Enable with env var on relay process: + # STRFRY_EVENT_PAYLOAD_ZSTD_LEVEL=3 + # Valid range: 1-22 (higher = more CPU, potentially better ratio) } relay { From 4e2fc4b5ff8aa9797aac9bf06aa18710ee0733d5 Mon Sep 17 00:00:00 2001 From: Strange Signal Date: Sat, 28 Feb 2026 20:57:38 -0500 Subject: [PATCH 2/3] feat: add STRFRY_RETAIN_REPLACED_EVENTS option for audit trail When enabled, strfry keeps ALL versions of replaceable events (kinds 0, 3, 10000-19999, 30000-39999) instead of deleting old versions when newer ones arrive. This is useful for: - Audit trails - Historical analysis - Compliance requirements Enable with env var: STRFRY_RETAIN_REPLACED_EVENTS=true --- src/events.cpp | 29 ++++++++++++++++++++++++++--- strfry.conf | 6 ++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/src/events.cpp b/src/events.cpp index 45a9599a..1bb7cc7f 100644 --- a/src/events.cpp +++ b/src/events.cpp @@ -23,6 +23,16 @@ static int getPayloadZstdLevel() { return level; } +static bool shouldRetainReplacedEvents() { + static bool retain = []() { + const char *s = std::getenv("STRFRY_RETAIN_REPLACED_EVENTS"); + if (!s) return false; + return std::string(s) == "1" || std::string(s) == "true"; + }(); + + return retain; +} + std::string nostrJsonToPackedEvent(const tao::json::value &v) { PackedEventTagBuilder tagBuilder; @@ -334,10 +344,23 @@ void writeEvents(lmdb::txn &txn, NegentropyFilterCache &neFilterCache, std::vect if (otherTimestamp < thisTimestamp || (otherTimestamp == thisTimestamp && packed.id() < otherPacked.id())) { - if (logLevel >= 1) LI << "Deleting event (d-tag). id=" << to_hex(otherPacked.id()); - levIdsToDelete.push_back(otherEv.primaryKeyId); + // New event is newer - normally we'd delete the old one + if (shouldRetainReplacedEvents()) { + // Audit mode: keep both versions, just log + if (logLevel >= 1) LI << "Retaining replaced event (audit mode). id=" << to_hex(otherPacked.id()); + } else { + if (logLevel >= 1) LI << "Deleting event (d-tag). id=" << to_hex(otherPacked.id()); + levIdsToDelete.push_back(otherEv.primaryKeyId); + } } else { - ev.status = EventWriteStatus::Replaced; + // New event is older - normally we'd reject it + if (shouldRetainReplacedEvents()) { + // Audit mode: keep both versions, store the older one too + if (logLevel >= 1) LI << "Storing older replaceable event (audit mode). id=" << to_hex(packed.id()); + // Don't set Replaced status - let it be stored + } else { + ev.status = EventWriteStatus::Replaced; + } } } diff --git a/strfry.conf b/strfry.conf index 3ce2c743..0769c3bf 100644 --- a/strfry.conf +++ b/strfry.conf @@ -42,6 +42,12 @@ events { # Enable with env var on relay process: # STRFRY_EVENT_PAYLOAD_ZSTD_LEVEL=3 # Valid range: 1-22 (higher = more CPU, potentially better ratio) + + # Optional: Retain replaced events for audit trail. + # By default, when a newer replaceable event (kind 0, 3, 10000-19999, 30000-39999) + # arrives, the old version is deleted. Enable this to keep ALL versions: + # STRFRY_RETAIN_REPLACED_EVENTS=true + # This creates a full audit trail but increases storage usage. } relay { From 6934c6579b8026614b38034d8ad3c135bbc2b435 Mon Sep 17 00:00:00 2001 From: Strange Signal Date: Sat, 28 Feb 2026 21:18:30 -0500 Subject: [PATCH 3/3] fix: safer STRFRY_RETAIN_REPLACED_EVENTS implementation The previous implementation could cause index corruption by storing older incoming events. This fix: 1. When NEWER event arrives: keep old event in storage (audit) but still update replace index to point to new event. NIP-01 queries return only latest; old events queryable by ID. 2. When OLDER event arrives: reject it (even in audit mode) to prevent index corruption. Use external JSONL audit log to capture these. This preserves NIP-01 compliance while enabling audit trail for events that were superseded by newer versions. --- src/events.cpp | 20 +++++++++----------- strfry.conf | 14 ++++++++++++-- 2 files changed, 21 insertions(+), 13 deletions(-) diff --git a/src/events.cpp b/src/events.cpp index 1bb7cc7f..982a4cac 100644 --- a/src/events.cpp +++ b/src/events.cpp @@ -344,23 +344,21 @@ void writeEvents(lmdb::txn &txn, NegentropyFilterCache &neFilterCache, std::vect if (otherTimestamp < thisTimestamp || (otherTimestamp == thisTimestamp && packed.id() < otherPacked.id())) { - // New event is newer - normally we'd delete the old one + // New event is newer - delete the old one (unless audit mode) if (shouldRetainReplacedEvents()) { - // Audit mode: keep both versions, just log - if (logLevel >= 1) LI << "Retaining replaced event (audit mode). id=" << to_hex(otherPacked.id()); + // Audit mode: keep old event in storage for historical queries. + // The replace index will be updated to point to the new event, + // so NIP-01 queries return only the latest version. + // Old event remains queryable by ID for audit purposes. + if (logLevel >= 1) LI << "Retaining replaced event in storage (audit mode). id=" << to_hex(otherPacked.id()); } else { if (logLevel >= 1) LI << "Deleting event (d-tag). id=" << to_hex(otherPacked.id()); levIdsToDelete.push_back(otherEv.primaryKeyId); } } else { - // New event is older - normally we'd reject it - if (shouldRetainReplacedEvents()) { - // Audit mode: keep both versions, store the older one too - if (logLevel >= 1) LI << "Storing older replaceable event (audit mode). id=" << to_hex(packed.id()); - // Don't set Replaced status - let it be stored - } else { - ev.status = EventWriteStatus::Replaced; - } + // New event is older - reject it (even in audit mode, to avoid index corruption). + // The JSONL audit log in the ingestion layer captures these anyway. + ev.status = EventWriteStatus::Replaced; } } diff --git a/strfry.conf b/strfry.conf index 0769c3bf..92989843 100644 --- a/strfry.conf +++ b/strfry.conf @@ -45,9 +45,19 @@ events { # Optional: Retain replaced events for audit trail. # By default, when a newer replaceable event (kind 0, 3, 10000-19999, 30000-39999) - # arrives, the old version is deleted. Enable this to keep ALL versions: + # arrives, the old version is deleted from storage. + # + # Enable this to KEEP old versions in storage: # STRFRY_RETAIN_REPLACED_EVENTS=true - # This creates a full audit trail but increases storage usage. + # + # Behavior when enabled: + # - Old versions are kept in storage (queryable by event ID) + # - Replace index still points to newest (NIP-01 queries return latest only) + # - Older incoming events are still rejected (to prevent index corruption) + # - Storage usage increases over time + # + # For complete audit including rejected older events, use an external + # JSONL audit log in your ingestion layer. } relay {