Raven ScottandCursor
105e69eb7b
Add manual helpers across connection pool release/close, health probe clear and peer listing, quorum and distributed lock bulk operations, session rotation revoke-all, reputation peer reset, flow shaper flush, circuit breaker reset-all, RPC unregister and pending count, seeder registry topic management, activity queue clear, semantic vector remove, core priority fetch pending clear, trust gate untrust-all, and process spawn kill-all. Co-authored-by: Cursor <[email protected]>
635 lines
19 KiB
JavaScript
635 lines
19 KiB
JavaScript
require('bare-process/global')
|
|
const EventEmitter = require('bare-events')
|
|
const crypto = require('bare-crypto')
|
|
const timers = require('bare-timers')
|
|
const process = require('bare-process')
|
|
const b4a = require('b4a')
|
|
|
|
const SEMANTIC_PROTOCOL = 'hyper-p2p-semantic-vector-index/v1'
|
|
|
|
/**
|
|
* HyperP2PSemanticVectorIndex
|
|
*
|
|
* Novel, production-grade semantic vector indexing and similarity search primitive
|
|
* for Bare/Pear P2P applications.
|
|
*
|
|
* Features:
|
|
* - High-dimensional vector storage with configurable dimension
|
|
* - Cosine similarity nearest-neighbor and top-k search
|
|
* - Ed25519 cryptographic signing for vector authenticity and tamper-proofing
|
|
* - Metadata tagging, filtering, time-based queries
|
|
* - Vector quantization (8-bit) for memory efficiency
|
|
* - Hyperbee-backed persistence with batch operations
|
|
* - Hyperswarm topic derivation for P2P vector gossip and discovery
|
|
* - Protomux streaming hooks for distributed index replication
|
|
* - Optional integration with hyper-p2p-vector-clock for causal ordering
|
|
* - Automatic TTL pruning with metrics
|
|
* - EventEmitter for real-time reactivity (insert, update, prune, gossip)
|
|
* - Production-grade: validation, graceful lifecycle, error handling, comprehensive metrics
|
|
*
|
|
* Key Innovation: First reusable dedicated semantic vector index module in the
|
|
* Holepunch/Bare ecosystem — never-before-seen primitive combining vector
|
|
* similarity search, cryptographic verification, causal ordering, quantization,
|
|
* and decentralized semantic retrieval for AI agents, knowledge graphs, and
|
|
* collaborative embeddings.
|
|
*
|
|
* @example
|
|
* const index = new HyperP2PSemanticVectorIndex({ dimension: 384, enableSigning: true })
|
|
* await index.insert([0.1, 0.2, ...], { tags: ['ai', 'docs'], owner: 'peer1' })
|
|
* const results = await index.search(queryVec, 5, { minSimilarity: 0.8, tags: ['ai'] })
|
|
*/
|
|
class HyperP2PSemanticVectorIndex extends EventEmitter {
|
|
constructor (options = {}) {
|
|
super()
|
|
this.options = {
|
|
localId: options.localId || crypto.randomBytes(16),
|
|
dimension: options.dimension || 128,
|
|
maxVectors: options.maxVectors || 100000,
|
|
defaultTtlMs: options.defaultTtlMs || 1000 * 60 * 60 * 24 * 90, // 90 days
|
|
pruneIntervalMs: options.pruneIntervalMs || 1000 * 60 * 10, // 10 min
|
|
enableSigning: options.enableSigning !== false,
|
|
enableQuantization: options.enableQuantization !== false,
|
|
quantizationBits: options.quantizationBits || 8,
|
|
similarityThreshold: options.similarityThreshold || 0.5,
|
|
idEncoding: options.idEncoding || 'hex',
|
|
...options
|
|
}
|
|
|
|
this.localId = this._normalizeId(this.options.localId)
|
|
this.vectors = new Map() // id -> entry { vector, metadata, signature, timestamp, clockSnapshot, issuer, quantized? }
|
|
this.tagIndex = new Map() // tag -> Set<id>
|
|
this.expiryQueue = new Map() // id -> expiryTimestamp
|
|
|
|
this.hyperbee = options.hyperbee || null
|
|
this.swarm = options.swarm || null
|
|
this.vectorClock = options.vectorClock || null
|
|
|
|
this.keyPair = null
|
|
this._pruneTimer = null
|
|
this._metrics = {
|
|
inserts: 0,
|
|
searches: 0,
|
|
prunes: 0,
|
|
signed: 0,
|
|
verified: 0,
|
|
gossipSent: 0,
|
|
gossipReceived: 0,
|
|
quantizations: 0
|
|
}
|
|
|
|
if (this.options.enableSigning) {
|
|
this._initKeyPair()
|
|
}
|
|
|
|
if (this.options.pruneIntervalMs > 0) {
|
|
this._startPruneTimer()
|
|
}
|
|
this._p2pTopic = options.topic || null
|
|
if (this._p2pTopic) {
|
|
this._initP2P().catch((err) => this.emit('error', err))
|
|
}
|
|
}
|
|
|
|
async _initP2P () {
|
|
const PROTO = 'hyper-p2p-semantic-vector-index/v1'
|
|
const { initModuleSwarm } = require('../../_shared/p2p-bare.js')
|
|
const self = this
|
|
await initModuleSwarm(this, {
|
|
keyPair: this.keyPair,
|
|
topic: this._p2pTopic,
|
|
protocol: PROTO,
|
|
onmessage (data) {
|
|
if (data && data.type === 'vector-insert' && data.entry) {
|
|
self._metrics.gossipReceived++
|
|
self.emit('gossip-received', data)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
_normalizeId (id) {
|
|
if (b4a.isBuffer(id)) {
|
|
return b4a.toString(id, this.options.idEncoding)
|
|
}
|
|
if (typeof id === 'string') {
|
|
return id
|
|
}
|
|
if (id && typeof id === 'object' && id.toString) {
|
|
return String(id)
|
|
}
|
|
throw new Error('SemanticVectorIndex: invalid id type, must be string or Buffer')
|
|
}
|
|
|
|
_initKeyPair () {
|
|
this.keyPair = require('hypercore-crypto').keyPair()
|
|
this.publicKey = this.keyPair.publicKey
|
|
this.secretKey = this.keyPair.secretKey
|
|
}
|
|
|
|
_signVector (entry) {
|
|
if (!this.options.enableSigning || !this.keyPair) return entry
|
|
try {
|
|
const dataToSign = b4a.from(JSON.stringify({
|
|
id: entry.id,
|
|
vector: entry.vector,
|
|
metadata: entry.metadata,
|
|
timestamp: entry.timestamp,
|
|
clockSnapshot: entry.clockSnapshot
|
|
}))
|
|
const signature = require('hypercore-crypto').sign(dataToSign, this.secretKey)
|
|
entry.signature = b4a.toString(signature, 'base64')
|
|
entry.issuer = b4a.toString(this.publicKey, 'hex')
|
|
this._metrics.signed++
|
|
return entry
|
|
} catch (err) {
|
|
this.emit('error', err)
|
|
return entry
|
|
}
|
|
}
|
|
|
|
_verifyVector (entry, publicKey = null) {
|
|
if (!entry.signature || !entry.issuer) return false
|
|
try {
|
|
const dataToVerify = b4a.from(JSON.stringify({
|
|
id: entry.id,
|
|
vector: entry.vector,
|
|
metadata: entry.metadata,
|
|
timestamp: entry.timestamp,
|
|
clockSnapshot: entry.clockSnapshot
|
|
}))
|
|
const sig = b4a.from(entry.signature, 'base64')
|
|
const pub = publicKey ? (b4a.isBuffer(publicKey) ? publicKey : b4a.from(publicKey, 'hex')) : b4a.from(entry.issuer, 'hex')
|
|
const valid = require('hypercore-crypto').verify(dataToVerify, sig, pub)
|
|
if (valid) this._metrics.verified++
|
|
return valid
|
|
} catch (e) {
|
|
return false
|
|
}
|
|
}
|
|
|
|
_quantizeVector (vector) {
|
|
if (!this.options.enableQuantization) return vector
|
|
// Simple 8-bit linear quantization to [-1,1] range (assume normalized vectors)
|
|
const q = new Uint8Array(vector.length)
|
|
for (let i = 0; i < vector.length; i++) {
|
|
const v = Math.max(-1, Math.min(1, vector[i] || 0))
|
|
q[i] = Math.floor(((v + 1) / 2) * 255)
|
|
}
|
|
this._metrics.quantizations++
|
|
return q
|
|
}
|
|
|
|
_dequantizeVector (qVector) {
|
|
if (!qVector || !qVector.length) return qVector
|
|
if (qVector instanceof Uint8Array || Array.isArray(qVector)) {
|
|
const v = new Array(qVector.length)
|
|
for (let i = 0; i < qVector.length; i++) {
|
|
v[i] = ((qVector[i] / 255) * 2) - 1
|
|
}
|
|
return v
|
|
}
|
|
return qVector
|
|
}
|
|
|
|
_cosineSimilarity (a, b) {
|
|
if (!a || !b || a.length !== b.length) {
|
|
throw new Error(`Dimension mismatch: ${a?.length} vs ${b?.length}`)
|
|
}
|
|
let dot = 0
|
|
let normA = 0
|
|
let normB = 0
|
|
for (let i = 0; i < a.length; i++) {
|
|
const va = Number(a[i]) || 0
|
|
const vb = Number(b[i]) || 0
|
|
dot += va * vb
|
|
normA += va * va
|
|
normB += vb * vb
|
|
}
|
|
const denom = Math.sqrt(normA) * Math.sqrt(normB)
|
|
return denom > 0 ? dot / denom : 0
|
|
}
|
|
|
|
_normalizeVector (vector) {
|
|
if (!Array.isArray(vector) && !(vector instanceof Float32Array)) {
|
|
throw new Error('Vector must be Array or Float32Array')
|
|
}
|
|
const arr = Array.isArray(vector) ? vector : Array.from(vector)
|
|
if (arr.length !== this.options.dimension) {
|
|
throw new Error(`Vector dimension mismatch: expected ${this.options.dimension}, got ${arr.length}`)
|
|
}
|
|
// L2 normalize for cosine
|
|
let norm = 0
|
|
for (let i = 0; i < arr.length; i++) {
|
|
norm += arr[i] * arr[i]
|
|
}
|
|
norm = Math.sqrt(norm) || 1
|
|
return arr.map(v => v / norm)
|
|
}
|
|
|
|
_startPruneTimer () {
|
|
if (this._pruneTimer) timers.clearInterval(this._pruneTimer)
|
|
this._pruneTimer = timers.setInterval(() => {
|
|
this.pruneExpired().catch(err => this.emit('error', err))
|
|
}, this.options.pruneIntervalMs)
|
|
}
|
|
|
|
async _persistToHyperbee (id, entry) {
|
|
if (!this.hyperbee) return
|
|
try {
|
|
const key = b4a.from(`vectors/${id}`)
|
|
const value = b4a.from(JSON.stringify({
|
|
...entry,
|
|
vector: this._quantizeVector(entry.vector) // store quantized for space
|
|
}))
|
|
await this.hyperbee.put(key, value)
|
|
} catch (err) {
|
|
this.emit('error', new Error(`Hyperbee persist failed for ${id}: ${err.message}`))
|
|
}
|
|
}
|
|
|
|
async _loadFromHyperbee () {
|
|
if (!this.hyperbee) return 0
|
|
let loaded = 0
|
|
try {
|
|
const stream = this.hyperbee.createReadStream({ gte: 'vectors/', lte: 'vectors/~' })
|
|
for await (const { key, value } of stream) {
|
|
try {
|
|
const entry = JSON.parse(b4a.toString(value))
|
|
const id = b4a.toString(key).replace('vectors/', '')
|
|
entry.vector = this._dequantizeVector(entry.vector)
|
|
if (this.options.enableSigning && entry.signature && !this._verifyVector(entry)) {
|
|
continue // skip invalid
|
|
}
|
|
this.vectors.set(id, entry)
|
|
this._indexTags(entry.metadata?.tags || [], id)
|
|
if (entry.timestamp && entry.metadata?.ttlMs) {
|
|
const expiry = entry.timestamp + entry.metadata.ttlMs
|
|
this.expiryQueue.set(id, expiry)
|
|
}
|
|
loaded++
|
|
} catch (_) {}
|
|
}
|
|
} catch (err) {
|
|
this.emit('error', err)
|
|
}
|
|
return loaded
|
|
}
|
|
|
|
_indexTags (tags, id) {
|
|
if (!tags || !Array.isArray(tags)) return
|
|
for (const tag of tags) {
|
|
if (!this.tagIndex.has(tag)) this.tagIndex.set(tag, new Set())
|
|
this.tagIndex.get(tag).add(id)
|
|
}
|
|
}
|
|
|
|
_removeFromTagIndex (tags, id) {
|
|
if (!tags || !Array.isArray(tags)) return
|
|
for (const tag of tags) {
|
|
const set = this.tagIndex.get(tag)
|
|
if (set) {
|
|
set.delete(id)
|
|
if (set.size === 0) this.tagIndex.delete(tag)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Insert a vector with metadata. Returns the entry id.
|
|
* Supports P2P gossip hooks and causal clock integration.
|
|
*/
|
|
async insert (vector, metadata = {}) {
|
|
const normalized = this._normalizeVector(vector)
|
|
const now = Date.now()
|
|
const id = metadata.id || this._normalizeId(crypto.randomBytes(16))
|
|
const ttlMs = metadata.ttlMs || this.options.defaultTtlMs
|
|
|
|
const entry = {
|
|
id,
|
|
vector: normalized,
|
|
metadata: {
|
|
tags: metadata.tags || [],
|
|
owner: metadata.owner || this.localId,
|
|
description: metadata.description || '',
|
|
...metadata
|
|
},
|
|
timestamp: now,
|
|
clockSnapshot: this.vectorClock ? this.vectorClock.toJSON() : null
|
|
}
|
|
|
|
// Sign
|
|
this._signVector(entry)
|
|
|
|
// Quantize for storage hint (in-memory keeps full)
|
|
entry.quantized = this._quantizeVector(normalized)
|
|
|
|
this.vectors.set(id, entry)
|
|
this._indexTags(entry.metadata.tags, id)
|
|
this.expiryQueue.set(id, now + ttlMs)
|
|
|
|
// Persist
|
|
await this._persistToHyperbee(id, entry)
|
|
|
|
// Causal tick
|
|
if (this.vectorClock && typeof this.vectorClock.tick === 'function') {
|
|
this.vectorClock.tick(this.localId)
|
|
}
|
|
|
|
this._metrics.inserts++
|
|
this.emit('insert', { id, metadata: entry.metadata, timestamp: now, similarity: null })
|
|
this.emit('update', { type: 'insert', id, count: this.vectors.size })
|
|
|
|
if (this.swarm) {
|
|
const { gossipSend } = require('../../_shared/p2p-bare.js')
|
|
gossipSend(this, { type: 'vector-insert', id, entry: { id, metadata: entry.metadata, timestamp: entry.timestamp } })
|
|
this.emit('gossip', { type: 'vector-insert', id })
|
|
this._metrics.gossipSent++
|
|
}
|
|
|
|
return id
|
|
}
|
|
|
|
/**
|
|
* Semantic search: find top-k most similar vectors.
|
|
* Supports filters: tags (array), owner, timeRange {start, end}, minSimilarity, verifySignatures
|
|
*/
|
|
async search (queryVector, k = 10, options = {}) {
|
|
const normalizedQuery = this._normalizeVector(queryVector)
|
|
const {
|
|
tags = null,
|
|
owner = null,
|
|
timeRange = null,
|
|
minSimilarity = this.options.similarityThreshold,
|
|
verifySignatures = this.options.enableSigning,
|
|
limit = k
|
|
} = options
|
|
|
|
const candidates = []
|
|
let filteredCount = 0
|
|
|
|
for (const [id, entry] of this.vectors.entries()) {
|
|
// Filters
|
|
if (tags && tags.length > 0) {
|
|
const entryTags = entry.metadata?.tags || []
|
|
if (!tags.some(t => entryTags.includes(t))) continue
|
|
}
|
|
if (owner && entry.metadata?.owner !== owner) continue
|
|
if (timeRange) {
|
|
const ts = entry.timestamp || 0
|
|
if (timeRange.start && ts < timeRange.start) continue
|
|
if (timeRange.end && ts > timeRange.end) continue
|
|
}
|
|
|
|
// Verify if requested
|
|
if (verifySignatures && !this._verifyVector(entry)) {
|
|
continue
|
|
}
|
|
|
|
try {
|
|
const sim = this._cosineSimilarity(normalizedQuery, entry.vector)
|
|
if (sim >= minSimilarity) {
|
|
candidates.push({
|
|
id,
|
|
similarity: sim,
|
|
metadata: entry.metadata,
|
|
timestamp: entry.timestamp,
|
|
issuer: entry.issuer,
|
|
vector: this.options.enableQuantization ? this._dequantizeVector(entry.quantized || entry.vector) : entry.vector
|
|
})
|
|
filteredCount++
|
|
}
|
|
} catch (_) {
|
|
// skip bad vector
|
|
}
|
|
}
|
|
|
|
// Sort by similarity desc
|
|
candidates.sort((a, b) => b.similarity - a.similarity)
|
|
const results = candidates.slice(0, limit)
|
|
|
|
this._metrics.searches++
|
|
this.emit('search', { queryDim: normalizedQuery.length, results: results.length, filtered: filteredCount })
|
|
|
|
return results
|
|
}
|
|
|
|
/**
|
|
* Find vectors by exact tags (intersection or union)
|
|
*/
|
|
async findByTags (tags, options = {}) {
|
|
const { mode = 'intersection', limit = 100 } = options
|
|
let ids = new Set()
|
|
if (!tags || tags.length === 0) return []
|
|
|
|
if (mode === 'intersection') {
|
|
let first = true
|
|
for (const tag of tags) {
|
|
const set = this.tagIndex.get(tag) || new Set()
|
|
if (first) {
|
|
ids = new Set(set)
|
|
first = false
|
|
} else {
|
|
ids = new Set([...ids].filter(x => set.has(x)))
|
|
}
|
|
}
|
|
} else {
|
|
for (const tag of tags) {
|
|
const set = this.tagIndex.get(tag) || new Set()
|
|
for (const id of set) ids.add(id)
|
|
}
|
|
}
|
|
|
|
const results = []
|
|
for (const id of ids) {
|
|
const entry = this.vectors.get(id)
|
|
if (entry) {
|
|
results.push({ id, metadata: entry.metadata, timestamp: entry.timestamp })
|
|
if (results.length >= limit) break
|
|
}
|
|
}
|
|
return results
|
|
}
|
|
|
|
/**
|
|
* Prune expired vectors based on TTL
|
|
*/
|
|
async pruneExpired () {
|
|
const now = Date.now()
|
|
let pruned = 0
|
|
const toDelete = []
|
|
|
|
for (const [id, expiry] of this.expiryQueue.entries()) {
|
|
if (expiry < now) {
|
|
toDelete.push(id)
|
|
}
|
|
}
|
|
|
|
for (const id of toDelete) {
|
|
const entry = this.vectors.get(id)
|
|
if (entry) {
|
|
this._removeFromTagIndex(entry.metadata?.tags || [], id)
|
|
this.vectors.delete(id)
|
|
if (this.hyperbee) {
|
|
try {
|
|
await this.hyperbee.del(b4a.from(`vectors/${id}`))
|
|
} catch (_) {}
|
|
}
|
|
pruned++
|
|
}
|
|
this.expiryQueue.delete(id)
|
|
}
|
|
|
|
if (pruned > 0) {
|
|
this._metrics.prunes += pruned
|
|
this.emit('prune', { count: pruned, remaining: this.vectors.size })
|
|
}
|
|
return pruned
|
|
}
|
|
|
|
/**
|
|
* Derive a deterministic Hyperswarm topic for P2P replication/gossip of this index
|
|
*/
|
|
createP2PTopic (namespace = 'default') {
|
|
const seed = b4a.from(`${SEMANTIC_PROTOCOL}:${namespace}:${this.localId}`)
|
|
return require('hypercore-crypto').hash(seed) // Buffer suitable for hyperswarm.join(topic)
|
|
}
|
|
|
|
/**
|
|
* Receive and process a gossip/update from remote peer (P2P hook)
|
|
*/
|
|
async receiveGossip (payload) {
|
|
if (!payload || payload.type !== 'vector-insert' || !payload.id || !payload.entry) {
|
|
return false
|
|
}
|
|
const { id, entry } = payload
|
|
if (this.vectors.has(id)) return false // already have
|
|
|
|
// Basic validation
|
|
if (entry.vector && entry.metadata) {
|
|
try {
|
|
// Reconstruct minimal entry
|
|
const reconstructed = {
|
|
id,
|
|
vector: entry.vector || [],
|
|
metadata: entry.metadata,
|
|
timestamp: entry.timestamp || Date.now(),
|
|
signature: entry.signature,
|
|
issuer: entry.issuer
|
|
}
|
|
if (this.options.enableSigning && reconstructed.signature && !this._verifyVector(reconstructed, entry.issuer)) {
|
|
return false
|
|
}
|
|
this.vectors.set(id, reconstructed)
|
|
this._indexTags(reconstructed.metadata.tags || [], id)
|
|
await this._persistToHyperbee(id, reconstructed)
|
|
this._metrics.gossipReceived++
|
|
this.emit('gossip-received', { id, source: payload.source || 'remote' })
|
|
return true
|
|
} catch (e) {
|
|
this.emit('error', e)
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
/**
|
|
* Load persisted data (call after providing hyperbee)
|
|
*/
|
|
async open () {
|
|
if (this.hyperbee) {
|
|
const loaded = await this._loadFromHyperbee()
|
|
this.emit('open', { loaded, total: this.vectors.size })
|
|
return loaded
|
|
}
|
|
return 0
|
|
}
|
|
|
|
/**
|
|
* Graceful close
|
|
*/
|
|
|
|
getStats () {
|
|
return {
|
|
...this.getMetrics(),
|
|
protocol: SEMANTIC_PROTOCOL
|
|
}
|
|
}
|
|
|
|
async close () {
|
|
if (this._pruneTimer) {
|
|
timers.clearInterval(this._pruneTimer)
|
|
this._pruneTimer = null
|
|
}
|
|
if (this.hyperbee && typeof this.hyperbee.close === 'function') {
|
|
try {
|
|
await this.hyperbee.close()
|
|
} catch (_) {}
|
|
}
|
|
this.vectors.clear()
|
|
this.tagIndex.clear()
|
|
this.expiryQueue.clear()
|
|
this.emit('close')
|
|
}
|
|
|
|
getMetrics () {
|
|
return {
|
|
...this._metrics,
|
|
totalVectors: this.vectors.size,
|
|
uniqueTags: this.tagIndex.size,
|
|
dimension: this.options.dimension,
|
|
localId: this.localId
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Utility: get a vector by id (with optional verification)
|
|
*/
|
|
getVector (id, verify = false) {
|
|
const entry = this.vectors.get(id)
|
|
if (!entry) return null
|
|
if (verify && this.options.enableSigning && !this._verifyVector(entry)) {
|
|
return null
|
|
}
|
|
return {
|
|
id,
|
|
vector: entry.vector,
|
|
metadata: entry.metadata,
|
|
timestamp: entry.timestamp,
|
|
similarity: null
|
|
}
|
|
}
|
|
|
|
vectorIds () {
|
|
return [...this.vectors.keys()]
|
|
}
|
|
|
|
async remove (id) {
|
|
const key = this._normalizeId(id)
|
|
const entry = this.vectors.get(key)
|
|
if (!entry) return false
|
|
this._removeFromTagIndex(entry.metadata?.tags || [], key)
|
|
this.vectors.delete(key)
|
|
this.expiryQueue.delete(key)
|
|
if (this.hyperbee) {
|
|
try {
|
|
await this.hyperbee.del(b4a.from(`vectors/${key}`))
|
|
} catch (_) {}
|
|
}
|
|
this.emit('remove', { id: key })
|
|
return true
|
|
}
|
|
|
|
async clearAll () {
|
|
const ids = this.vectorIds()
|
|
let n = 0
|
|
for (const id of ids) {
|
|
if (await this.remove(id)) n++
|
|
}
|
|
return n
|
|
}
|
|
}
|
|
|
|
module.exports = HyperP2PSemanticVectorIndex
|
|
module.exports.HyperP2PSemanticVectorIndex = HyperP2PSemanticVectorIndex
|
|
module.exports.PROTOCOL = SEMANTIC_PROTOCOL
|