From 48f452e63f9522bddff834bc41d5f93a3fc13cae Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Sun, 11 Oct 2026 00:40:32 +0530 Subject: [PATCH 1/4] feat(nip77): add negentropy set reconciliation core Signed-off-by: Priyanshubhartistm --- .changeset/nip77-negentropy-core.md | 5 + .knip.json | 1 + src/utils/negentropy.ts | 532 ++++++++++++++++++++++++++++ 3 files changed, 538 insertions(+) create mode 100644 .changeset/nip77-negentropy-core.md create mode 100644 src/utils/negentropy.ts diff --git a/.changeset/nip77-negentropy-core.md b/.changeset/nip77-negentropy-core.md new file mode 100644 index 000000000..da6ad0e68 --- /dev/null +++ b/.changeset/nip77-negentropy-core.md @@ -0,0 +1,5 @@ +--- +"nostream": minor +--- + +feat(nip77): add negentropy set-reconciliation core and a sorted event id query diff --git a/.knip.json b/.knip.json index b42e6a95b..4151e7b65 100644 --- a/.knip.json +++ b/.knip.json @@ -15,6 +15,7 @@ ], "ignore": [ ".nostr/**", + "src/utils/negentropy.ts", "src/utils/relay-probe/**" ], "commitlint": false, diff --git a/src/utils/negentropy.ts b/src/utils/negentropy.ts new file mode 100644 index 000000000..df951e0d5 --- /dev/null +++ b/src/utils/negentropy.ts @@ -0,0 +1,532 @@ +// Negentropy (NIP-77) set reconciliation, protocol version 1. +// +// TypeScript port of the reference implementation at https://github.com/hoytech/negentropy +// (js/Negentropy.js), (C) 2023 Doug Hoyte, MIT license. The wire format is unchanged; the port is +// synchronous (server-side only) and bounds-checks all input read from the network. + +import { createHash } from 'crypto' + +export const NEGENTROPY_PROTOCOL_VERSION = 0x61 + +const ID_SIZE = 32 +const FINGERPRINT_SIZE = 16 +const MIN_FRAME_SIZE_LIMIT = 4096 +// Headroom left under frameSizeLimit for the trailing fingerprint bound the reference appends. +const FRAME_SIZE_HEADROOM = 200 +const BUCKETS = 16 +const MAX_VARINT_BYTES = 8 + +// Sentinel for the "infinity" upper bound. It is encoded as 0 on the wire and never as a number. +export const MAX_TIMESTAMP = Number.MAX_SAFE_INTEGER + +enum Mode { + Skip = 0, + Fingerprint = 1, + IdList = 2, +} + +export interface NegentropyItem { + timestamp: number + id: Uint8Array +} + +export interface ReconcileResult { + // Next message to send to the peer, or null when the initiator has nothing left to send. + output: Uint8Array | null + // Hex ids we have and the peer lacks / the peer has and we lack. Only filled in for the initiator. + haveIds: string[] + needIds: string[] +} + +const toHex = (bytes: Uint8Array): string => + Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength).toString('hex') + +const compareBytes = (a: Uint8Array, b: Uint8Array): number => { + const shared = Math.min(a.byteLength, b.byteLength) + for (let i = 0; i < shared; i++) { + if (a[i] !== b[i]) { + return a[i] < b[i] ? -1 : 1 + } + } + + return a.byteLength < b.byteLength ? -1 : a.byteLength > b.byteLength ? 1 : 0 +} + +const compareItems = (a: NegentropyItem, b: NegentropyItem): number => + a.timestamp === b.timestamp ? compareBytes(a.id, b.id) : a.timestamp < b.timestamp ? -1 : 1 + +const encodeVarInt = (n: number): Uint8Array => { + if (!Number.isSafeInteger(n) || n < 0) { + throw new Error('varint out of range') + } + if (n === 0) { + return Uint8Array.of(0) + } + + const groups: number[] = [] + for (let rest = n; rest > 0; rest = Math.floor(rest / 128)) { + groups.push(rest % 128) + } + groups.reverse() + for (let i = 0; i < groups.length - 1; i++) { + groups[i] |= 128 + } + + return Uint8Array.from(groups) +} + +class ByteWriter { + private readonly chunks: Uint8Array[] = [] + private size = 0 + + public get length(): number { + return this.size + } + + public write(bytes: Uint8Array): void { + this.chunks.push(bytes) + this.size += bytes.byteLength + } + + public writeVarInt(n: number): void { + this.write(encodeVarInt(n)) + } + + public toBytes(): Uint8Array { + return Buffer.concat(this.chunks, this.size) + } +} + +class ByteReader { + private offset = 0 + + public constructor(private readonly buffer: Uint8Array) {} + + public get remaining(): number { + return this.buffer.byteLength - this.offset + } + + public readByte(): number { + return this.readBytes(1)[0] + } + + public readBytes(n: number): Uint8Array { + if (n > this.remaining) { + throw new Error('parse ends prematurely') + } + const bytes = this.buffer.subarray(this.offset, this.offset + n) + this.offset += n + + return bytes + } + + public readVarInt(): number { + let result = 0 + for (let i = 0; i < MAX_VARINT_BYTES; i++) { + const byte = this.readByte() + result = result * 128 + (byte & 127) + if ((byte & 128) === 0) { + if (!Number.isSafeInteger(result)) { + throw new Error('varint out of range') + } + + return result + } + } + + throw new Error('varint too long') + } +} + +export class NegentropyStorageVector { + private items: NegentropyItem[] = [] + private sealed = false + + public insert(timestamp: number, id: Uint8Array): void { + if (this.sealed) { + throw new Error('already sealed') + } + if (!Number.isSafeInteger(timestamp) || timestamp < 0 || timestamp >= MAX_TIMESTAMP) { + throw new Error('bad timestamp for added item') + } + if (id.byteLength !== ID_SIZE) { + throw new Error('bad id size for added item') + } + + this.items.push({ timestamp, id }) + } + + public seal(): void { + if (this.sealed) { + throw new Error('already sealed') + } + + this.items.sort(compareItems) + for (let i = 1; i < this.items.length; i++) { + if (compareItems(this.items[i - 1], this.items[i]) === 0) { + throw new Error('duplicate item inserted') + } + } + this.sealed = true + } + + public size(): number { + this.assertSealed() + + return this.items.length + } + + public iterate(begin: number, end: number, callback: (item: NegentropyItem, index: number) => boolean): void { + this.assertSealed() + this.assertBounds(begin, end) + + for (let i = begin; i < end; i++) { + if (!callback(this.items[i], i)) { + break + } + } + } + + // Index of the first item in [begin, end) that is >= bound, or end if there is none. + public findLowerBound(begin: number, end: number, bound: NegentropyItem): number { + this.assertSealed() + this.assertBounds(begin, end) + + let first = begin + let count = end - begin + while (count > 0) { + const step = Math.floor(count / 2) + const middle = first + step + if (compareItems(this.items[middle], bound) < 0) { + first = middle + 1 + count -= step + 1 + } else { + count = step + } + } + + return first + } + + // Sum of the ids in [begin, end) as a 256-bit little-endian integer mod 2^256, hashed together + // with the item count; the first 16 bytes of the SHA-256 are the fingerprint. + public fingerprint(begin: number, end: number): Uint8Array { + this.assertSealed() + this.assertBounds(begin, end) + + const sum = new DataView(new ArrayBuffer(ID_SIZE)) + for (let i = begin; i < end; i++) { + const id = new DataView(this.items[i].id.buffer, this.items[i].id.byteOffset, ID_SIZE) + let carry = 0 + for (let offset = 0; offset < ID_SIZE; offset += 4) { + const total = sum.getUint32(offset, true) + id.getUint32(offset, true) + carry + sum.setUint32(offset, total >>> 0, true) + carry = total > 0xffffffff ? 1 : 0 + } + } + + const hash = createHash('sha256') + .update(new Uint8Array(sum.buffer)) + .update(encodeVarInt(end - begin)) + .digest() + + return hash.subarray(0, FINGERPRINT_SIZE) + } + + private assertSealed(): void { + if (!this.sealed) { + throw new Error('not sealed') + } + } + + private assertBounds(begin: number, end: number): void { + if (begin > end || end > this.items.length) { + throw new Error('bad range') + } + } +} + +export class Negentropy { + private isInitiator = false + private lastTimestampIn = 0 + private lastTimestampOut = 0 + + public constructor( + private readonly storage: NegentropyStorageVector, + private readonly frameSizeLimit = 0, + ) { + if (frameSizeLimit !== 0 && frameSizeLimit < MIN_FRAME_SIZE_LIMIT) { + throw new Error('frameSizeLimit too small') + } + } + + // Client side only: builds the first message of a session. + public initiate(): Uint8Array { + if (this.isInitiator) { + throw new Error('already initiated') + } + this.isInitiator = true + this.resetTimestamps() + + const output = new ByteWriter() + output.write(Uint8Array.of(NEGENTROPY_PROTOCOL_VERSION)) + this.splitRange(0, this.storage.size(), { timestamp: MAX_TIMESTAMP, id: new Uint8Array(0) }, output) + + return output.toBytes() + } + + public reconcile(message: Uint8Array): ReconcileResult { + const haveIds: string[] = [] + const needIds: string[] = [] + const query = new ByteReader(message) + + // Timestamps are delta-encoded per message. + this.resetTimestamps() + + const fullOutput = new ByteWriter() + fullOutput.write(Uint8Array.of(NEGENTROPY_PROTOCOL_VERSION)) + + const protocolVersion = query.readByte() + if (protocolVersion < 0x60 || protocolVersion > 0x6f) { + throw new Error('invalid negentropy protocol version byte') + } + if (protocolVersion !== NEGENTROPY_PROTOCOL_VERSION) { + if (this.isInitiator) { + throw new Error(`unsupported negentropy protocol version requested: ${protocolVersion - 0x60}`) + } + + // Tell the peer which version we speak and let it retry. + return { output: fullOutput.toBytes(), haveIds, needIds } + } + + const storageSize = this.storage.size() + let prevBound: NegentropyItem = { timestamp: 0, id: new Uint8Array(0) } + let prevIndex = 0 + let skip = false + + while (query.remaining !== 0) { + let o = new ByteWriter() + + const doSkip = () => { + if (skip) { + skip = false + this.encodeBound(prevBound, o) + o.writeVarInt(Mode.Skip) + } + } + + const currBound = this.decodeBound(query) + const mode = query.readVarInt() + + const lower = prevIndex + let upper = this.storage.findLowerBound(prevIndex, storageSize, currBound) + + if (mode === Mode.Skip) { + skip = true + } else if (mode === Mode.Fingerprint) { + const theirFingerprint = query.readBytes(FINGERPRINT_SIZE) + const ourFingerprint = this.storage.fingerprint(lower, upper) + + if (compareBytes(theirFingerprint, ourFingerprint) !== 0) { + doSkip() + this.splitRange(lower, upper, currBound, o) + } else { + skip = true + } + } else if (mode === Mode.IdList) { + const numIds = query.readVarInt() + if (numIds * ID_SIZE > query.remaining) { + throw new Error('parse ends prematurely') + } + + const theirIds = new Map() + for (let i = 0; i < numIds; i++) { + const id = query.readBytes(ID_SIZE) + if (this.isInitiator) { + theirIds.set(toHex(id), id) + } + } + + if (this.isInitiator) { + skip = true + + this.storage.iterate(lower, upper, (item) => { + const key = toHex(item.id) + if (theirIds.has(key)) { + theirIds.delete(key) + } else { + haveIds.push(key) + } + + return true + }) + + for (const key of theirIds.keys()) { + needIds.push(key) + } + } else { + doSkip() + + const responseIds = new ByteWriter() + let numResponseIds = 0 + let endBound = currBound + + this.storage.iterate(lower, upper, (item, index) => { + if (this.exceededFrameSizeLimit(fullOutput.length + responseIds.length)) { + endBound = item + // Shrink upper so the remaining range gets the correct fingerprint below. + upper = index + return false + } + + responseIds.write(item.id) + numResponseIds++ + return true + }) + + this.encodeBound(endBound, o) + o.writeVarInt(Mode.IdList) + o.writeVarInt(numResponseIds) + o.write(responseIds.toBytes()) + + fullOutput.write(o.toBytes()) + o = new ByteWriter() + } + } else { + throw new Error('unexpected mode') + } + + if (this.exceededFrameSizeLimit(fullOutput.length + o.length)) { + // Stop range processing and answer with one fingerprint covering everything that is left. + const remainingFingerprint = this.storage.fingerprint(upper, storageSize) + + this.encodeBound({ timestamp: MAX_TIMESTAMP, id: new Uint8Array(0) }, fullOutput) + fullOutput.writeVarInt(Mode.Fingerprint) + fullOutput.write(remainingFingerprint) + break + } + + fullOutput.write(o.toBytes()) + + prevIndex = upper + prevBound = currBound + } + + return { + output: fullOutput.length === 1 && this.isInitiator ? null : fullOutput.toBytes(), + haveIds, + needIds, + } + } + + private splitRange(lower: number, upper: number, upperBound: NegentropyItem, o: ByteWriter): void { + const numElems = upper - lower + + if (numElems < BUCKETS * 2) { + this.encodeBound(upperBound, o) + o.writeVarInt(Mode.IdList) + o.writeVarInt(numElems) + this.storage.iterate(lower, upper, (item) => { + o.write(item.id) + return true + }) + + return + } + + const itemsPerBucket = Math.floor(numElems / BUCKETS) + const bucketsWithExtra = numElems % BUCKETS + let curr = lower + + for (let i = 0; i < BUCKETS; i++) { + const bucketSize = itemsPerBucket + (i < bucketsWithExtra ? 1 : 0) + const ourFingerprint = this.storage.fingerprint(curr, curr + bucketSize) + curr += bucketSize + + let nextBound: NegentropyItem + if (curr === upper) { + nextBound = upperBound + } else { + let prevItem: NegentropyItem | undefined + let currItem: NegentropyItem | undefined + this.storage.iterate(curr - 1, curr + 1, (item, index) => { + if (index === curr - 1) { + prevItem = item + } else { + currItem = item + } + + return true + }) + + nextBound = this.getMinimalBound(prevItem as NegentropyItem, currItem as NegentropyItem) + } + + this.encodeBound(nextBound, o) + o.writeVarInt(Mode.Fingerprint) + o.write(ourFingerprint) + } + } + + private exceededFrameSizeLimit(size: number): boolean { + return this.frameSizeLimit !== 0 && size > this.frameSizeLimit - FRAME_SIZE_HEADROOM + } + + private resetTimestamps(): void { + this.lastTimestampIn = 0 + this.lastTimestampOut = 0 + } + + private decodeTimestampIn(reader: ByteReader): number { + const encoded = reader.readVarInt() + if (encoded === 0 || this.lastTimestampIn === MAX_TIMESTAMP) { + this.lastTimestampIn = MAX_TIMESTAMP + return MAX_TIMESTAMP + } + + const timestamp = encoded - 1 + this.lastTimestampIn + if (!Number.isSafeInteger(timestamp) || timestamp >= MAX_TIMESTAMP) { + throw new Error('timestamp out of range') + } + this.lastTimestampIn = timestamp + + return timestamp + } + + private decodeBound(reader: ByteReader): NegentropyItem { + const timestamp = this.decodeTimestampIn(reader) + const length = reader.readVarInt() + if (length > ID_SIZE) { + throw new Error('bound key too long') + } + + return { timestamp, id: reader.readBytes(length) } + } + + private encodeBound(bound: NegentropyItem, o: ByteWriter): void { + if (bound.timestamp === MAX_TIMESTAMP) { + this.lastTimestampOut = MAX_TIMESTAMP + o.writeVarInt(0) + } else { + o.writeVarInt(bound.timestamp - this.lastTimestampOut + 1) + this.lastTimestampOut = bound.timestamp + } + + o.writeVarInt(bound.id.byteLength) + o.write(bound.id) + } + + // The shortest bound that still sorts after prev and at-or-before curr. + private getMinimalBound(prev: NegentropyItem, curr: NegentropyItem): NegentropyItem { + if (curr.timestamp !== prev.timestamp) { + return { timestamp: curr.timestamp, id: new Uint8Array(0) } + } + + let sharedPrefixBytes = 0 + while (sharedPrefixBytes < ID_SIZE && curr.id[sharedPrefixBytes] === prev.id[sharedPrefixBytes]) { + sharedPrefixBytes++ + } + + return { timestamp: curr.timestamp, id: curr.id.subarray(0, sharedPrefixBytes + 1) } + } +} From b70e3bddd35e733a218d78201f049b3b4f96d923 Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Sun, 11 Oct 2026 00:41:47 +0530 Subject: [PATCH 2/4] feat(nip77): add repository query for reconciliation items Signed-off-by: Priyanshubhartistm --- src/@types/repositories.ts | 3 +++ src/repositories/event-repository.ts | 33 +++++++++++++++++++++++++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index 087e8ea31..7d45cf102 100644 --- a/src/@types/repositories.ts +++ b/src/@types/repositories.ts @@ -31,6 +31,8 @@ export interface EventPurgeCounts { export type ExposedPromiseKeys = 'then' | 'catch' | 'finally' +export type ReconciliationItem = Pick + export interface IQueryResult extends Pick, keyof Promise & ExposedPromiseKeys> { stream(options?: Record): PassThrough & AsyncIterable } @@ -42,6 +44,7 @@ export interface IEventRepository { upsertMany(events: Event[]): Promise findByFilters(filters: SubscriptionFilter[]): IQueryResult countByFilters(filters: SubscriptionFilter[]): Promise + findReconciliationItems(filter: SubscriptionFilter): IQueryResult deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise deleteByPubkeyExceptKinds(pubkey: Pubkey, excludedKinds: number[]): Promise hasActiveRequestToVanish(pubkey: Pubkey): Promise diff --git a/src/repositories/event-repository.ts b/src/repositories/event-repository.ts index 3cf4eb70e..3ac583162 100644 --- a/src/repositories/event-repository.ts +++ b/src/repositories/event-repository.ts @@ -23,7 +23,13 @@ import { } from '../constants/base' import { DatabaseClient, EventId } from '../@types/base' import { DBEvent, Event } from '../@types/event' -import { EventPurgeCounts, EventRetentionOptions, IEventRepository, IQueryResult } from '../@types/repositories' +import { + EventPurgeCounts, + EventRetentionOptions, + IEventRepository, + IQueryResult, + ReconciliationItem, +} from '../@types/repositories' import { toBuffer, toJSON } from '../utils/transform' import { createLogger } from '../factories/logger-factory' import { isGenericTagQuery, isGeohashPrefixCriterion, stripGeohashPrefixWildcard } from '../utils/filter' @@ -128,6 +134,31 @@ export class EventRepository implements IEventRepository { return Number(result?.count ?? 0) } + /** + * Every live event matching the filter as (created_at, id), oldest first, for NIP-77 set + * reconciliation. The filter's limit is ignored on purpose: reconciliation compares whole sets, so + * truncating one side would report the cut-off events as missing. Deleted and expired events are + * excluded because they must never be advertised to a peer. + */ + public findReconciliationItems(filter: SubscriptionFilter): IQueryResult { + logger('querying reconciliation items for %o', filter) + + const now = Math.floor(Date.now() / 1000) + const builder = this.readReplicaDbClient('events').select('events.event_created_at', 'events.event_id') + + this.applyFilterConditions(builder, filter) + + builder + .whereNull('events.deleted_at') + .andWhere((bd) => { + bd.whereNull('events.expires_at').orWhere('events.expires_at', '>', now) + }) + .orderBy('events.event_created_at', 'asc') + .orderBy('events.event_id', 'asc') + + return builder as unknown as IQueryResult + } + // Wraps every query so each keeps its own ORDER BY and LIMIT. first.union(rest) would put the // first query's ORDER BY and LIMIT after the UNION, applying them to the whole result. private unionFilterQueries(queries: any[]): any { From 2289f42ff7b5a3e424cf2ad3bb5f229a72395f02 Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Sun, 11 Oct 2026 00:43:03 +0530 Subject: [PATCH 3/4] test(nip77): cover negentropy core and reconciliation query Signed-off-by: Priyanshubhartistm --- .../repositories/event-repository.spec.ts | 74 ++++ test/unit/utils/negentropy.spec.ts | 396 ++++++++++++++++++ 2 files changed, 470 insertions(+) create mode 100644 test/unit/utils/negentropy.spec.ts diff --git a/test/unit/repositories/event-repository.spec.ts b/test/unit/repositories/event-repository.spec.ts index 1bccee290..38c91e20a 100644 --- a/test/unit/repositories/event-repository.spec.ts +++ b/test/unit/repositories/event-repository.spec.ts @@ -684,6 +684,80 @@ describe('EventRepository', () => { }) }) + describe('.findReconciliationItems', () => { + const NOW_SECONDS = 1791658580 + const LIVE_EVENTS_CLAUSE = + '"events"."deleted_at" is null and ("events"."expires_at" is null or "events"."expires_at" > 1791658580)' + const ORDER_CLAUSE = 'order by "events"."event_created_at" asc, "events"."event_id" asc' + + beforeEach(() => { + sandbox.stub(Date, 'now').returns(NOW_SECONDS * 1000) + }) + + it('returns a function with stream and then', () => { + const result = repository.findReconciliationItems({}) + + expect(result).to.have.property('stream') + expect(result).to.have.property('then') + }) + + it('selects only created_at and id, oldest first, skipping deleted and expired events', () => { + const query = repository.findReconciliationItems({}).toString() + + expect(query).to.equal( + `select "events"."event_created_at", "events"."event_id" from "events" where ${LIVE_EVENTS_CLAUSE} ${ORDER_CLAUSE}`, + ) + }) + + it('applies the filter conditions', () => { + const query = repository + .findReconciliationItems({ kinds: [1], authors: ['a'.repeat(64)], since: 5, until: 9 }) + .toString() + + expect(query).to.include(`"event_pubkey" in (X'${'a'.repeat(64)}')`) + expect(query).to.include('"event_kind" in (1)') + expect(query).to.include('"event_created_at" >= 5') + expect(query).to.include('"event_created_at" <= 9') + expect(query).to.include(LIVE_EVENTS_CLAUSE) + }) + + it('ignores the filter limit, since reconciliation needs the whole matching set', () => { + const query = repository.findReconciliationItems({ kinds: [1], limit: 3 }).toString() + + expect(query).to.not.include('limit') + expect(query).to.include(ORDER_CLAUSE) + }) + + it('applies tag filters', () => { + const query = repository.findReconciliationItems({ '#e': ['b'.repeat(64)] } as any).toString() + + expect(query).to.include('event_tags') + expect(query).to.include(LIVE_EVENTS_CLAUSE) + }) + + it('excludes events matching an actionable report when hiding is enabled', () => { + const hideEnabledRepository = new EventRepository( + dbClient, + rrDbClient, + () => + ({ + nip56: { enabled: true, trustedModerators: [], hideActionableReports: true }, + }) as any, + ) + + const query = hideEnabledRepository.findReconciliationItems({ kinds: [1] }).toString() + + expect(query).to.include('not exists') + expect(query).to.include('"reports"."actionable" = true') + }) + + it('does not touch the reports table when hiding is disabled', () => { + const query = repository.findReconciliationItems({ kinds: [1] }).toString() + + expect(query).to.not.include('reports') + }) + }) + describe('.countByFilters', () => { it('throws error if filters is empty', async () => { try { diff --git a/test/unit/utils/negentropy.spec.ts b/test/unit/utils/negentropy.spec.ts new file mode 100644 index 000000000..fe3290fd9 --- /dev/null +++ b/test/unit/utils/negentropy.spec.ts @@ -0,0 +1,396 @@ +import { expect } from 'chai' +import { createHash } from 'crypto' + +import { MAX_TIMESTAMP, Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' + +const BASE_TIMESTAMP = 1700000000 + +const idFor = (label: string): Buffer => createHash('sha256').update(label).digest() + +type Entry = [timestamp: number, id: Buffer] + +const buildEntries = (from: number, to: number, perTimestamp = 4): Entry[] => { + const entries: Entry[] = [] + for (let i = from; i < to; i++) { + entries.push([BASE_TIMESTAMP + Math.floor(i / perTimestamp), idFor(`negentropy-test-${i}`)]) + } + + return entries +} + +const buildStorage = (entries: Entry[]): NegentropyStorageVector => { + const storage = new NegentropyStorageVector() + for (const [timestamp, id] of entries) { + storage.insert(timestamp, id) + } + storage.seal() + + return storage +} + +const hex = (bytes: Uint8Array | null): string => (bytes ? Buffer.from(bytes).toString('hex') : 'null') + +const idsOf = (entries: Entry[]): string[] => entries.map(([, id]) => id.toString('hex')) + +const difference = (left: Entry[], right: Entry[]): string[] => { + const rightIds = new Set(idsOf(right)) + + return idsOf(left) + .filter((id) => !rightIds.has(id)) + .sort() +} + +interface SyncOutcome { + rounds: number + haveIds: string[] + needIds: string[] +} + +// Runs a full session: `client` initiates, `server` only answers, until the client has nothing left to say. +const sync = (client: Entry[], server: Entry[], serverFrameSizeLimit = 0): SyncOutcome => { + const initiator = new Negentropy(buildStorage(client)) + const responder = new Negentropy(buildStorage(server), serverFrameSizeLimit) + // Under a frame size limit the same id can be reported again in a later round, exactly as in the + // reference implementation, so an initiator has to accumulate into a set. + const haveIds = new Set() + const needIds = new Set() + + let message: Uint8Array | null = initiator.initiate() + let rounds = 0 + while (message !== null) { + rounds++ + expect(rounds, 'session did not converge').to.be.lessThan(50) + + const reply = responder.reconcile(message).output + const result = initiator.reconcile(reply as Uint8Array) + result.haveIds.forEach((id) => haveIds.add(id)) + result.needIds.forEach((id) => needIds.add(id)) + message = result.output + } + + return { rounds, haveIds: [...haveIds].sort(), needIds: [...needIds].sort() } +} + +describe('negentropy', () => { + describe('NegentropyStorageVector', () => { + it('rejects ids that are not 32 bytes', () => { + const storage = new NegentropyStorageVector() + + expect(() => storage.insert(BASE_TIMESTAMP, Buffer.alloc(31))).to.throw('bad id size for added item') + }) + + it('rejects timestamps that are negative, fractional or the infinity sentinel', () => { + const storage = new NegentropyStorageVector() + + expect(() => storage.insert(-1, idFor('a'))).to.throw('bad timestamp') + expect(() => storage.insert(1.5, idFor('a'))).to.throw('bad timestamp') + expect(() => storage.insert(MAX_TIMESTAMP, idFor('a'))).to.throw('bad timestamp') + }) + + it('refuses to insert or seal twice once sealed', () => { + const storage = buildStorage(buildEntries(0, 2)) + + expect(() => storage.insert(BASE_TIMESTAMP, idFor('late'))).to.throw('already sealed') + expect(() => storage.seal()).to.throw('already sealed') + }) + + it('rejects duplicate items when sealing', () => { + const storage = new NegentropyStorageVector() + storage.insert(BASE_TIMESTAMP, idFor('dup')) + storage.insert(BASE_TIMESTAMP, idFor('dup')) + + expect(() => storage.seal()).to.throw('duplicate item inserted') + }) + + it('refuses to be read before it is sealed', () => { + const storage = new NegentropyStorageVector() + + expect(() => storage.size()).to.throw('not sealed') + expect(() => storage.fingerprint(0, 0)).to.throw('not sealed') + }) + + it('sorts by timestamp, then by id', () => { + const low = Buffer.alloc(32, 0x01) + const high = Buffer.alloc(32, 0x02) + const storage = new NegentropyStorageVector() + storage.insert(BASE_TIMESTAMP + 1, low) + storage.insert(BASE_TIMESTAMP, high) + storage.insert(BASE_TIMESTAMP, low) + storage.seal() + + const order: string[] = [] + storage.iterate(0, storage.size(), (item) => { + order.push(`${item.timestamp - BASE_TIMESTAMP}:${item.id[0]}`) + return true + }) + + expect(order).to.deep.equal(['0:1', '0:2', '1:1']) + }) + + it('finds the lower bound of a bound', () => { + const storage = buildStorage(buildEntries(0, 10, 1)) + + expect(storage.findLowerBound(0, 10, { timestamp: 0, id: new Uint8Array(0) })).to.equal(0) + expect(storage.findLowerBound(0, 10, { timestamp: BASE_TIMESTAMP + 3, id: new Uint8Array(0) })).to.equal(3) + expect(storage.findLowerBound(0, 10, { timestamp: BASE_TIMESTAMP + 99, id: new Uint8Array(0) })).to.equal(10) + expect(storage.findLowerBound(0, 10, { timestamp: MAX_TIMESTAMP, id: new Uint8Array(0) })).to.equal(10) + }) + + it('rejects out-of-range bounds', () => { + const storage = buildStorage(buildEntries(0, 3)) + + expect(() => storage.iterate(0, 4, () => true)).to.throw('bad range') + expect(() => storage.fingerprint(2, 1)).to.throw('bad range') + }) + + it('produces 16-byte fingerprints that do not depend on insertion order', () => { + const entries = buildEntries(0, 8) + const forward = buildStorage(entries) + const backward = buildStorage([...entries].reverse()) + + expect(forward.fingerprint(0, 8)).to.have.lengthOf(16) + expect(hex(forward.fingerprint(0, 8))).to.equal(hex(backward.fingerprint(0, 8))) + }) + + it('produces different fingerprints for different sets and for different counts', () => { + const storage = buildStorage(buildEntries(0, 8)) + const other = buildStorage(buildEntries(1, 9)) + + expect(hex(storage.fingerprint(0, 8))).to.not.equal(hex(other.fingerprint(0, 8))) + expect(hex(storage.fingerprint(0, 4))).to.not.equal(hex(storage.fingerprint(0, 5))) + }) + + it('wraps the 256-bit id sum around, carrying across every 32-bit word', () => { + const allOnes = Buffer.alloc(32, 0xff) + const one = Buffer.alloc(32) + one[0] = 1 + const zero = Buffer.alloc(32) + + const overflowing = new NegentropyStorageVector() + overflowing.insert(BASE_TIMESTAMP, allOnes) + overflowing.insert(BASE_TIMESTAMP, one) + overflowing.seal() + + const zeroSum = new NegentropyStorageVector() + zeroSum.insert(BASE_TIMESTAMP, zero) + zeroSum.insert(BASE_TIMESTAMP + 1, zero) + zeroSum.seal() + + // (2^256 - 1) + 1 wraps to exactly 0 only if the carry ripples through all eight words. + expect(hex(overflowing.fingerprint(0, 2))).to.equal(hex(zeroSum.fingerprint(0, 2))) + }) + }) + + describe('wire compatibility with the reference implementation', () => { + // Generated with hoytech/negentropy js/Negentropy.js: client holds items 0-39, server holds 10-49. + const REFERENCE_FIRST_MESSAGE = + '6186aacfe20101c4019519545952534ac1dbf6efee31f306a50201bf010e45e53bc3ac3b96527fcd6c09d2047802017101744f77934468e967eeeabc0f798d8ef4020001873c27d9520afdf1a90c94fc3418b485010165011eb5a27d5fd29d201a8804d82fe10c2e02018e013ee24cc071884f88d81e56c138672c4402015801ce08fb40009f4496c99e002ce58159b402000166784dbdc5e2e720329752a4d7fe1fdc0101ca01e53cc691284a4fa5d32d61bf920bd03f020001392bfa2b9c1ea6c1e44575ee2cd47a9f0101cd01161b325b1e6b1f8c0e5e6dcce95c673a020001db0d53fffe5622719353467e5a4064340101a101f1c06ede4cff82a242fda6253d18a06f020001d5785181bb433c6f0d3ff48de323251a01019c01a054d5da1b42babccc1272c167eb56db0000015dfae5a6e074b6e1d2014f2b9134c08f' + const REFERENCE_RESPONSE = + '6186aacfe20101c402000201bf02000201710200020002029103607c71902fd449f5737b2dda7c5a4ee31e0cf1fb7cf13f2f951d3b131c169266fae12bbe176c85e6cc8940ff97afc4b7f899f13192cbd756856aaa06261f07019c000000020c9c185f5a0cb59b837e6ffc631a537507a7eefede982d9663577ae545fab5b287d0f8bae1791c4ccb49a9c267107c73043df98a26bc696febebd84c864beb68160f39aca329fab189af2e828cd50a518ac803854a3ae5c280b1e3873cdefc268e609a8d39e4370faed4290168ca6c9a7237f0e29923b0e48b45eb5ae6851bca6f8120620c1dfeb40c48a28f4e2dfe67c3f0cf490d80602e30a383700fb85cedf68d7f261ad6c9ff45c56880496fb12b1a1209c701ad76fbca9684a3490e1467a0689fb4d021cd5fa4ba91eda8283b7daddccd70fe41e26e6a502c17de020038ba6e665739195287c00becd4b248675c994d7c10f4b5a12c524b9b62e308740e31e05b1a766d5ba1ebf8ba17bee7da0164907525d095b6f18fbc3dd5b8197c1e11e56cdae96afbdba7c20dab5ddf6d34daa1847b809f79ada60945b8305d51448a6c8109848821eae5957bf56c1c8ca51b10d06cbbfda2ebcdbc8f5163a2cf6d5ddc25b6dde85fbcdb2e25ee0ec80c7815bf9784c8a710c66536a094bff846333d' + + it('encodes the same first message as the reference', () => { + const client = new Negentropy(buildStorage(buildEntries(0, 40))) + + expect(hex(client.initiate())).to.equal(REFERENCE_FIRST_MESSAGE) + }) + + it('answers that message exactly like the reference', () => { + const server = new Negentropy(buildStorage(buildEntries(10, 50))) + + const { output } = server.reconcile(Buffer.from(REFERENCE_FIRST_MESSAGE, 'hex')) + + expect(hex(output)).to.equal(REFERENCE_RESPONSE) + }) + + it('encodes an empty set as a single empty id list', () => { + const client = new Negentropy(buildStorage([])) + + expect(hex(client.initiate())).to.equal('6100000200') + }) + }) + + describe('reconciliation', () => { + it('finds nothing to sync between identical sets', () => { + const entries = buildEntries(0, 200) + + const outcome = sync(entries, entries) + + expect(outcome.haveIds).to.deep.equal([]) + expect(outcome.needIds).to.deep.equal([]) + expect(outcome.rounds).to.equal(1) + }) + + it('reports everything as needed when the client has nothing', () => { + const server = buildEntries(0, 50) + + const outcome = sync([], server) + + expect(outcome.haveIds).to.deep.equal([]) + expect(outcome.needIds).to.deep.equal(idsOf(server).sort()) + }) + + it('reports everything as had when the server has nothing', () => { + const client = buildEntries(0, 50) + + const outcome = sync(client, []) + + expect(outcome.haveIds).to.deep.equal(idsOf(client).sort()) + expect(outcome.needIds).to.deep.equal([]) + }) + + it('handles completely disjoint sets', () => { + const client = buildEntries(0, 100) + const server = buildEntries(1000, 1100) + + const outcome = sync(client, server) + + expect(outcome.haveIds).to.deep.equal(idsOf(client).sort()) + expect(outcome.needIds).to.deep.equal(idsOf(server).sort()) + }) + + for (const [clientSize, serverSize, serverOffset] of [ + [500, 480, 10], + [3000, 2900, 100], + [40, 5000, 0], + ]) { + it(`converges on the exact difference for ${clientSize} vs ${serverSize} items`, () => { + const client = buildEntries(0, clientSize) + const server = buildEntries(0, serverSize).slice(serverOffset) + + const outcome = sync(client, server) + + expect(outcome.haveIds).to.deep.equal(difference(client, server)) + expect(outcome.needIds).to.deep.equal(difference(server, client)) + }) + } + + it('converges when thousands of events share one timestamp', () => { + const client = buildEntries(0, 1500, 1500) + const server = buildEntries(100, 1600, 1500) + + const outcome = sync(client, server) + + expect(outcome.haveIds).to.deep.equal(difference(client, server)) + expect(outcome.needIds).to.deep.equal(difference(server, client)) + }) + + it('converges under a small frame size limit, in more rounds', () => { + // Large differences force the responder to cut its answer short and defer the rest. + const client = buildEntries(0, 3000) + const server = buildEntries(5000, 8000) + + const unlimited = sync(client, server) + const limited = sync(client, server, 4096) + + expect(limited.haveIds).to.deep.equal(difference(client, server)) + expect(limited.needIds).to.deep.equal(difference(server, client)) + expect(limited.rounds).to.be.greaterThan(unlimited.rounds) + }) + + it('keeps every responder message within the frame size limit', () => { + const frameSizeLimit = 4096 + const initiator = new Negentropy(buildStorage(buildEntries(0, 3000))) + const responder = new Negentropy(buildStorage(buildEntries(5000, 8000)), frameSizeLimit) + + let message: Uint8Array | null = initiator.initiate() + while (message !== null) { + const reply = responder.reconcile(message).output as Uint8Array + expect(reply.byteLength).to.be.at.most(frameSizeLimit) + message = initiator.reconcile(reply).output + } + }) + + it('rejects frame size limits below 4096', () => { + expect(() => new Negentropy(buildStorage([]), 4095)).to.throw('frameSizeLimit too small') + expect(() => new Negentropy(buildStorage([]), 0)).to.not.throw() + }) + + it('refuses to initiate twice', () => { + const client = new Negentropy(buildStorage([])) + client.initiate() + + expect(() => client.initiate()).to.throw('already initiated') + }) + + it('can run several independent sessions over the same storage', () => { + const storage = buildStorage(buildEntries(10, 50)) + const first = new Negentropy(storage) + const second = new Negentropy(storage) + const message = new Negentropy(buildStorage(buildEntries(0, 40))).initiate() + + expect(hex(first.reconcile(message).output)).to.equal(hex(second.reconcile(message).output)) + }) + }) + + describe('malformed input', () => { + const responder = () => new Negentropy(buildStorage(buildEntries(0, 100))) + + it('rejects an empty message', () => { + expect(() => responder().reconcile(new Uint8Array(0))).to.throw('parse ends prematurely') + }) + + it('rejects a version byte outside 0x60-0x6f', () => { + expect(() => responder().reconcile(Uint8Array.of(0x42))).to.throw('invalid negentropy protocol version byte') + }) + + it('answers an unsupported but well-formed version with just its own version byte', () => { + const { output } = responder().reconcile(Uint8Array.of(0x62)) + + expect(hex(output)).to.equal('61') + }) + + it('throws when the initiator is answered with an unsupported version', () => { + const initiator = new Negentropy(buildStorage([])) + initiator.initiate() + + expect(() => initiator.reconcile(Uint8Array.of(0x62))).to.throw('unsupported negentropy protocol version') + }) + + it('rejects an unknown mode', () => { + // bound(timestamp 0 -> infinity, empty id), mode 7 + expect(() => responder().reconcile(Uint8Array.of(0x61, 0x00, 0x00, 0x07))).to.throw('unexpected mode') + }) + + it('rejects a bound id longer than 32 bytes', () => { + expect(() => responder().reconcile(Uint8Array.of(0x61, 0x00, 33))).to.throw('bound key too long') + }) + + it('rejects a truncated fingerprint', () => { + expect(() => responder().reconcile(Uint8Array.of(0x61, 0x00, 0x00, 0x01, 0xaa))).to.throw( + 'parse ends prematurely', + ) + }) + + it('rejects an id list that claims more ids than the message carries', () => { + // mode 2 (id list), count 100, no ids follow + expect(() => responder().reconcile(Uint8Array.of(0x61, 0x00, 0x00, 0x02, 100))).to.throw('parse ends prematurely') + }) + + it('rejects an id list whose declared size would overflow when multiplied out', () => { + const huge = Uint8Array.of(0x61, 0x00, 0x00, 0x02, 0xff, 0xff, 0xff, 0xff, 0x7f) + + expect(() => responder().reconcile(huge)).to.throw('parse ends prematurely') + }) + + it('rejects varints that never terminate', () => { + const endless = Uint8Array.of(0x61, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80) + + expect(() => responder().reconcile(endless)).to.throw('varint too long') + }) + + it('rejects varints beyond the safe integer range', () => { + const tooBig = Uint8Array.of(0x61, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x7f) + + expect(() => responder().reconcile(tooBig)).to.throw('varint out of range') + }) + + it('does not mutate its storage when the input is rejected', () => { + const storage = buildStorage(buildEntries(0, 100)) + const before = hex(storage.fingerprint(0, 100)) + + expect(() => new Negentropy(storage).reconcile(Uint8Array.of(0x61, 0x00, 33))).to.throw() + + expect(hex(storage.fingerprint(0, 100))).to.equal(before) + }) + }) +}) From 120e01b706efdb54fb18104e2f80d14acfa89f7e Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Sun, 11 Oct 2026 01:51:49 +0530 Subject: [PATCH 4/4] fix(nip77): address review findings in negentropy core Signed-off-by: Priyanshubhartistm --- src/utils/negentropy.ts | 28 +++++++-- test/unit/utils/negentropy.spec.ts | 92 ++++++++++++++++++++++++++++++ 2 files changed, 114 insertions(+), 6 deletions(-) diff --git a/src/utils/negentropy.ts b/src/utils/negentropy.ts index df951e0d5..dc8483cac 100644 --- a/src/utils/negentropy.ts +++ b/src/utils/negentropy.ts @@ -26,8 +26,8 @@ enum Mode { } export interface NegentropyItem { - timestamp: number - id: Uint8Array + readonly timestamp: number + readonly id: Readonly } export interface ReconcileResult { @@ -142,7 +142,7 @@ export class NegentropyStorageVector { private items: NegentropyItem[] = [] private sealed = false - public insert(timestamp: number, id: Uint8Array): void { + public insert(timestamp: number, id: Readonly): void { if (this.sealed) { throw new Error('already sealed') } @@ -153,7 +153,9 @@ export class NegentropyStorageVector { throw new Error('bad id size for added item') } - this.items.push({ timestamp, id }) + // Copy the id and freeze the item: sealing relies on the sort order and uniqueness checked in + // seal(), which a caller mutating its own buffer afterwards would silently break. + this.items.push(Object.freeze({ timestamp, id: Uint8Array.from(id) })) } public seal(): void { @@ -240,7 +242,13 @@ export class NegentropyStorageVector { } private assertBounds(begin: number, end: number): void { - if (begin > end || end > this.items.length) { + if ( + !Number.isSafeInteger(begin) || + !Number.isSafeInteger(end) || + begin < 0 || + begin > end || + end > this.items.length + ) { throw new Error('bad range') } } @@ -302,6 +310,10 @@ export class Negentropy { const storageSize = this.storage.size() let prevBound: NegentropyItem = { timestamp: 0, id: new Uint8Array(0) } let prevIndex = 0 + // Index of the first item not yet accounted for by anything written to fullOutput. The peer + // compares a trailing fingerprint against everything after the last bound we wrote, so that is + // where a truncated reply must restart, not at the end of the range whose answer was dropped. + let writtenUpTo = 0 let skip = false while (query.remaining !== 0) { @@ -390,6 +402,7 @@ export class Negentropy { o.write(responseIds.toBytes()) fullOutput.write(o.toBytes()) + writtenUpTo = upper o = new ByteWriter() } } else { @@ -398,7 +411,7 @@ export class Negentropy { if (this.exceededFrameSizeLimit(fullOutput.length + o.length)) { // Stop range processing and answer with one fingerprint covering everything that is left. - const remainingFingerprint = this.storage.fingerprint(upper, storageSize) + const remainingFingerprint = this.storage.fingerprint(writtenUpTo, storageSize) this.encodeBound({ timestamp: MAX_TIMESTAMP, id: new Uint8Array(0) }, fullOutput) fullOutput.writeVarInt(Mode.Fingerprint) @@ -407,6 +420,9 @@ export class Negentropy { } fullOutput.write(o.toBytes()) + if (o.length > 0) { + writtenUpTo = upper + } prevIndex = upper prevBound = currBound diff --git a/test/unit/utils/negentropy.spec.ts b/test/unit/utils/negentropy.spec.ts index fe3290fd9..f0461690e 100644 --- a/test/unit/utils/negentropy.spec.ts +++ b/test/unit/utils/negentropy.spec.ts @@ -28,6 +28,19 @@ const buildStorage = (entries: Entry[]): NegentropyStorageVector => { return storage } +const varint = (n: number): number[] => { + const groups: number[] = [] + for (let rest = n; rest > 0; rest = Math.floor(rest / 128)) { + groups.push(rest % 128) + } + if (groups.length === 0) { + groups.push(0) + } + groups.reverse() + + return groups.map((group, i) => (i < groups.length - 1 ? group | 128 : group)) +} + const hex = (bytes: Uint8Array | null): string => (bytes ? Buffer.from(bytes).toString('hex') : 'null') const idsOf = (entries: Entry[]): string[] => entries.map(([, id]) => id.toString('hex')) @@ -143,6 +156,47 @@ describe('negentropy', () => { expect(() => storage.fingerprint(2, 1)).to.throw('bad range') }) + it('rejects range indexes that are negative, fractional or not finite', () => { + const storage = buildStorage(buildEntries(0, 3)) + + for (const [begin, end] of [ + [-1, 0], + [0, 1.5], + [Number.NaN, 2], + [0, Number.POSITIVE_INFINITY], + [-1, 2], + ]) { + expect(() => storage.iterate(begin, end, () => true), `iterate(${begin}, ${end})`).to.throw('bad range') + expect(() => storage.fingerprint(begin, end), `fingerprint(${begin}, ${end})`).to.throw('bad range') + expect(() => storage.findLowerBound(begin, end, { timestamp: 0, id: new Uint8Array(0) })).to.throw('bad range') + } + }) + + it('is not affected by the caller changing an id buffer after inserting it', () => { + const id = Buffer.from(idFor('mutable')) + const storage = new NegentropyStorageVector() + storage.insert(BASE_TIMESTAMP, id) + storage.seal() + const before = hex(storage.fingerprint(0, 1)) + + id.fill(0xff) + + expect(hex(storage.fingerprint(0, 1))).to.equal(before) + }) + + it('hands out frozen items so callbacks cannot reorder a sealed set', () => { + const storage = buildStorage(buildEntries(0, 2)) + + storage.iterate(0, 2, (item) => { + expect(Object.isFrozen(item)).to.equal(true) + expect(() => { + ;(item as any).timestamp = 0 + }).to.throw(TypeError) + + return true + }) + }) + it('produces 16-byte fingerprints that do not depend on insertion order', () => { const entries = buildEntries(0, 8) const forward = buildStorage(entries) @@ -300,6 +354,44 @@ describe('negentropy', () => { } }) + it('restarts a frame-truncated reply at the last written bound, not after the dropped range', () => { + // The responder holds 160 events. The peer asks for the first 120 as an id list (3.8 KB of + // answers, almost the whole 4096-byte frame), then sends a mismatching fingerprint for the + // other 40. That second answer no longer fits and is dropped, so the trailing fingerprint + // must cover those 40 events: the peer compares it against everything after the last bound + // that was actually written, and a fingerprint that skipped them would match a peer with + // nothing there, ending the session without ever reporting the 40 events. + const entries = buildEntries(0, 160, 1) + const storage = buildStorage(entries) + const message = Uint8Array.from([ + 0x61, + ...varint(BASE_TIMESTAMP + 120 + 1), // upper bound of the first range, delta-encoded + 0, // bound id length + 2, // mode: id list + 0, // peer lists no ids + 0, // upper bound: infinity + 0, // bound id length + 1, // mode: fingerprint + ...new Array(16).fill(0), // a fingerprint that cannot match + ]) + + const reply = new Negentropy(storage, 4096).reconcile(message).output as Uint8Array + + const trailingFingerprint = hex(reply.subarray(reply.length - 16)) + expect(trailingFingerprint).to.equal(hex(storage.fingerprint(120, 160))) + expect(trailingFingerprint).to.not.equal(hex(storage.fingerprint(160, 160))) + }) + + it('still converges when several mismatched ranges are cut by the frame limit', () => { + const client = buildEntries(0, 60, 1) + const server = buildEntries(0, 3000, 1) + + const outcome = sync(client, server, 4096) + + expect(outcome.haveIds).to.deep.equal(difference(client, server)) + expect(outcome.needIds).to.deep.equal(difference(server, client)) + }) + it('rejects frame size limits below 4096', () => { expect(() => new Negentropy(buildStorage([]), 4095)).to.throw('frameSizeLimit too small') expect(() => new Negentropy(buildStorage([]), 0)).to.not.throw()