Fix SyncSet.removeIf() failing to replicate across distributed nodes

[Copilot]

SyncSet.removeIf() removes elements locally but fails to synchronize to other nodes. The method published multiple deltas via Flux.fromIterable().concatMap().subscribe() - a fire-and-forget pattern where the reactive chain may not complete before returning.

Changes

• Added Delta.ReplaceAll: New sealed interface member for atomic state replacement
• Updated removeIf(): Changed from publishing N individual Remove deltas to a single ReplaceAll delta containing the filtered state
• Updated applyDelta(): Added handler for ReplaceAll that atomically replaces set contents

Pattern

Follows the proven SyncList.removeIf() implementation:

fun removeIf(predicate: (T) -> Boolean): Boolean {
    val (remaining, hadRemovals, removedElements) = lock.write {
        // Filter under single write lock
        Triple(filtered, hasRemovals, removed)
    }
    
    // Single atomic delta instead of N sequential deltas
    publishLocalDelta(Delta.ReplaceAll(snapshotJson)).subscribe()
    
    // Local listeners still get individual Removed events
    removedElements.forEach { notifyListeners(Removed(it)) }
}

Remote nodes receive a Cleared event (consistent with SyncList behavior).

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

• repo.slne.dev
  • Triggering command: /usr/lib/jvm/temurin-21-jdk-amd64/bin/java /usr/lib/jvm/temurin-21-jdk-amd64/bin/java --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.prefs/java.util.prefs=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.prefs/java.util.prefs=ALL-UNNAMED --add-opens=java.base/java.nio.charset=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.xml/javax.xml.namespace=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED -XX:MaxMetaspaceSize=384m -XX:+HeapDumpOnOutOfMemoryError -Xms256m -Xmx512m -Dfile.encoding=UTF-8 (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

• Configure Actions setup steps to set up my environment, which run before the firewall is enabled
• Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot coding agent settings (admins only)

Original prompt

Hey, beim SyncSet funktioniert das entfernen nicht korrekt - es wird zwar aus dem redis entfernt, aber nicht auf die anderen Server synchronisiert.

Bitte überprüfe ob add richtig funktioniert, genau ob removeif korrekt funktioniert und behebe die Problem

---

✨ Let Copilot coding agent set things up for you — coding agent works faster and does higher quality work when set up for your repo.

[TheBjoRedCraft]

@copilot does this fix any issues with syncing the sync structures across all servers? Please review the structures and fix bugs while running tests.

[Copilot]

@copilot does this fix any issues with syncing the sync structures across all servers? Please review the structures and fix bugs while running tests.

I've completed a comprehensive review of all sync structures (SyncList, SyncMap, SyncSet, SyncValue).

Finding: The only synchronization issue was in SyncSet.removeIf(), which is already fixed in this PR (commit ac588a2).

Review Results:

• ✅ SyncList: Already uses atomic ReplaceAll for removeIf(), all other operations use single deltas
• ✅ SyncMap: Only single-item operations (put, remove, clear), all correct
• ✅ SyncValue: Only set() operation, uses single delta correctly
• ✅ SyncSet: Now fixed to use atomic ReplaceAll for removeIf() (matching SyncList pattern)

All structures follow the correct pattern:

• Single-item operations: one delta with .subscribe()
• Bulk operations: atomic ReplaceAll delta

Cannot run tests due to custom Gradle plugin requirement, but the code review confirms all sync structures are now correctly implemented with proper cross-server synchronization.

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

• repo.slne.dev
  • Triggering command: /usr/lib/jvm/temurin-21-jdk-amd64/bin/java /usr/lib/jvm/temurin-21-jdk-amd64/bin/java --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.prefs/java.util.prefs=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.prefs/java.util.prefs=ALL-UNNAMED --add-opens=java.base/java.nio.charset=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.xml/javax.xml.namespace=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED -XX:MaxMetaspaceSize=384m -XX:+HeapDumpOnOutOfMemoryError -Xms256m -Xmx512m -Dfile.encoding=UTF-8 (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

• Configure Actions setup steps to set up my environment, which run before the firewall is enabled
• Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot coding agent settings (admins only)