feat(quartz): re-evaluate filter membership on addressable supersession
When a replaceable / addressable update arrives in handleInsert, the projection now re-runs Filter.match against the new event for every filter. A v2 that no longer matches a filter (e.g. tag list changed) is removed from that filter's set; a v2 that newly matches a filter joins it. If no filter retains the slot afterwards, it's fully dropped from byId / byAddress. Closes the previous gap where v2 of an addressable kept a stale filter membership inherited from v1. The slot's MutableStateFlow is still updated in place — collectors of that handle still see the content change before the membership update emits. The cap-eviction-with-cleanup logic was extracted into a small `admit` helper since the supersession path and the new-slot path both need it. For the common case (filter on `kinds + authors`, no tag/time constraints), v2 always still matches — the new branches are no-ops and the list reference stays stable, preserving the in-place update guarantee. The behaviour change kicks in for tag, time-window, or id-list filters. New test addressableUpdateDropsSlotWhenFilterStopsMatching: v1 has hashtag "nostr" and matches a `t = nostr` filter; v2 changes to "bitcoin"; the projection drops the slot. 18/18 projection tests pass. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
This commit is contained in:
+60
-29
@@ -71,12 +71,13 @@ sealed interface ProjectionState<out T : Event> {
|
|||||||
* list. Stable list reference while membership is unchanged.
|
* list. Stable list reference while membership is unchanged.
|
||||||
* - **In-place replaceable / addressable update**: same slot, new
|
* - **In-place replaceable / addressable update**: same slot, new
|
||||||
* [MutableStateFlow.value]. Only collectors of that handle
|
* [MutableStateFlow.value]. Only collectors of that handle
|
||||||
* re-render. The slot's sort key is frozen at insertion, so the
|
* re-render. The slot's sort key is frozen at insertion so the
|
||||||
* list does not reshuffle when the new version has a later
|
* list ordering doesn't reshuffle on a later `created_at`. Filter
|
||||||
* `created_at`. *Caveat: the slot stays in whatever filters
|
* membership *is* re-evaluated against the new event, so a v2
|
||||||
* matched the original version; if the new version no longer
|
* that no longer matches a filter (e.g. tag changed) drops out;
|
||||||
* matches a filter (e.g. its tags changed), the slot is not
|
* a v2 that newly matches another filter joins it. In the common
|
||||||
* re-evaluated and remains live.*
|
* case (filter on `kinds + authors`) v2 still matches and the
|
||||||
|
* list reference stays stable.
|
||||||
* - **Removal**: NIP-09, NIP-62, NIP-40 expiration, `delete(filter)`.
|
* - **Removal**: NIP-09, NIP-62, NIP-40 expiration, `delete(filter)`.
|
||||||
*
|
*
|
||||||
* Limit is **per-filter**: each filter retains at most its own
|
* Limit is **per-filter**: each filter retains at most its own
|
||||||
@@ -194,17 +195,37 @@ class EventStoreProjection<T : Event>(
|
|||||||
if (existing != null) {
|
if (existing != null) {
|
||||||
if (!supersedes(event, existing.flow.value)) return false
|
if (!supersedes(event, existing.flow.value)) return false
|
||||||
|
|
||||||
// Same address, new winner. Rekey byId from the
|
// Same address, new winner. Rekey byId, mutate
|
||||||
// previous event id to the new one and update the
|
// flow.value in place, then re-evaluate filter
|
||||||
// handle's value in place — list reference stays the
|
// membership against the new event — v2 may add
|
||||||
// same; only the handle's collectors re-render.
|
// matches or lose previously-matching filters.
|
||||||
val previousId = existing.flow.value.id
|
val previousId = existing.flow.value.id
|
||||||
if (previousId != event.id) {
|
if (previousId != event.id) {
|
||||||
byId.remove(previousId)
|
byId.remove(previousId)
|
||||||
byId[event.id] = existing
|
byId[event.id] = existing
|
||||||
}
|
}
|
||||||
existing.flow.value = event as T
|
existing.flow.value = event as T
|
||||||
return false
|
|
||||||
|
var changed = false
|
||||||
|
for ((f, set) in perFilter) {
|
||||||
|
val nowMatches = f.match(event)
|
||||||
|
val wasIn = set.contains(existing)
|
||||||
|
when {
|
||||||
|
nowMatches && !wasIn -> {
|
||||||
|
if (admit(existing, f, set)) changed = true
|
||||||
|
}
|
||||||
|
|
||||||
|
!nowMatches && wasIn -> {
|
||||||
|
set.remove(existing)
|
||||||
|
changed = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// If no filter retains the slot anymore, fully drop it.
|
||||||
|
if (perFilter.values.none { it.contains(existing) }) {
|
||||||
|
if (removeIndexes(existing)) changed = true
|
||||||
|
}
|
||||||
|
return changed
|
||||||
}
|
}
|
||||||
} else if (byId.containsKey(event.id)) {
|
} else if (byId.containsKey(event.id)) {
|
||||||
return false
|
return false
|
||||||
@@ -213,32 +234,42 @@ class EventStoreProjection<T : Event>(
|
|||||||
// Genuinely new slot. Offer it to every matching filter; if
|
// Genuinely new slot. Offer it to every matching filter; if
|
||||||
// any filter still holds it after cap-eviction, the slot
|
// any filter still holds it after cap-eviction, the slot
|
||||||
// becomes live and gets indexed.
|
// becomes live and gets indexed.
|
||||||
var membershipChanged = false
|
var changed = false
|
||||||
val slot = Slot(event as T)
|
val slot = Slot(event as T)
|
||||||
for ((f, set) in perFilter) {
|
for ((f, set) in perFilter) {
|
||||||
if (!f.match(event)) continue
|
if (!f.match(event)) continue
|
||||||
set.add(slot)
|
if (admit(slot, f, set)) changed = true
|
||||||
val cap = f.limit ?: continue
|
|
||||||
while (set.size > cap) {
|
|
||||||
val tail = set.last()
|
|
||||||
set.remove(tail)
|
|
||||||
if (tail !== slot && perFilter.values.none { it.contains(tail) }) {
|
|
||||||
// Tail no longer retained by any filter — fully drop.
|
|
||||||
if (removeIndexes(tail)) membershipChanged = true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// The slot survived cap-eviction in at least one filter, so it
|
|
||||||
// belongs in the indexes. Otherwise nothing was indexed and
|
|
||||||
// the only membership effect is whatever evictions happened
|
|
||||||
// along the way.
|
|
||||||
if (perFilter.values.any { it.contains(slot) }) {
|
if (perFilter.values.any { it.contains(slot) }) {
|
||||||
byId[event.id] = slot
|
byId[event.id] = slot
|
||||||
if (address != null) byAddress[address] = slot
|
if (address != null) byAddress[address] = slot
|
||||||
membershipChanged = true
|
changed = true
|
||||||
}
|
}
|
||||||
return membershipChanged
|
return changed
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Add [slot] to filter [f]'s [set] and evict the tail if the cap
|
||||||
|
* is exceeded. Returns true if an eviction triggered a slot drop
|
||||||
|
* from the indexes (its only filter retention was the evicted
|
||||||
|
* one).
|
||||||
|
*/
|
||||||
|
private fun admit(
|
||||||
|
slot: Slot<T>,
|
||||||
|
f: Filter,
|
||||||
|
set: java.util.SortedSet<Slot<T>>,
|
||||||
|
): Boolean {
|
||||||
|
set.add(slot)
|
||||||
|
val cap = f.limit ?: return false
|
||||||
|
var changed = false
|
||||||
|
while (set.size > cap) {
|
||||||
|
val tail = set.last()
|
||||||
|
set.remove(tail)
|
||||||
|
if (tail !== slot && perFilter.values.none { it.contains(tail) }) {
|
||||||
|
if (removeIndexes(tail)) changed = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return changed
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun handleDeletion(deletion: DeletionEvent): Boolean {
|
private fun handleDeletion(deletion: DeletionEvent): Boolean {
|
||||||
|
|||||||
+44
@@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
|||||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.ObservableEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.ObservableEventStore
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.tags.hashtags.hashtag
|
||||||
import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent
|
import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent
|
||||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||||
import com.vitorpamplona.quartz.nip23LongContent.LongTextNoteEvent
|
import com.vitorpamplona.quartz.nip23LongContent.LongTextNoteEvent
|
||||||
@@ -208,6 +209,49 @@ class EventStoreProjectionTest {
|
|||||||
assertSame(seedList, projection.items, "addressable update must not change list reference")
|
assertSame(seedList, projection.items, "addressable update must not change list reference")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* If v2 of an addressable no longer matches the filter (e.g.
|
||||||
|
* tag list changed), the slot is dropped. The projection
|
||||||
|
* re-evaluates filter membership on every supersession.
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
fun addressableUpdateDropsSlotWhenFilterStopsMatching() =
|
||||||
|
runBlocking {
|
||||||
|
val time = TimeUtils.now()
|
||||||
|
// v1 carries tag "nostr"; v2 changes the tag to "bitcoin".
|
||||||
|
val v1 =
|
||||||
|
signer.sign(
|
||||||
|
LongTextNoteEvent.build("blog v1", "title", dTag = "blog", createdAt = time) {
|
||||||
|
hashtag("nostr")
|
||||||
|
},
|
||||||
|
)
|
||||||
|
observable.insert(v1)
|
||||||
|
|
||||||
|
// Filter narrows to events that ALSO carry hashtag "nostr".
|
||||||
|
val projection =
|
||||||
|
projectionOf<LongTextNoteEvent>(
|
||||||
|
Filter(
|
||||||
|
kinds = listOf(LongTextNoteEvent.KIND),
|
||||||
|
authors = listOf(v1.pubKey),
|
||||||
|
tags = mapOf("t" to listOf("nostr")),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
projection.awaitLoaded()
|
||||||
|
assertEquals(1, projection.items.size)
|
||||||
|
|
||||||
|
val v2 =
|
||||||
|
signer.sign(
|
||||||
|
LongTextNoteEvent.build("blog v2", "title", dTag = "blog", createdAt = time + 1) {
|
||||||
|
hashtag("bitcoin")
|
||||||
|
},
|
||||||
|
)
|
||||||
|
observable.insert(v2)
|
||||||
|
|
||||||
|
// v2 doesn't match — slot drops and the snapshot is empty.
|
||||||
|
val after = projection.awaitItems { it.isEmpty() }
|
||||||
|
assertTrue(after.isEmpty())
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Out-of-order arrival: the projection sees the *newer* version
|
* Out-of-order arrival: the projection sees the *newer* version
|
||||||
* first (e.g. the relay sent v2 first), then v1. The projection
|
* first (e.g. the relay sent v2 first), then v1. The projection
|
||||||
|
|||||||
Reference in New Issue
Block a user