GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Detect an unsubscribed observer by what the cache still holds, and pin the empty admission

Browse source 
A disposed R3 subscription ignores OnNext, so a no-op removal from the cache's
observer array cannot be seen by what a receiver gets; the test now checks the
cache stops holding the subscriber. An admission that changed nothing (a store
attached with no records) is not journaled.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 19:47:38 +02:00
commit 5b3885a1e8
2 changed files with 28 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

8
tests/GameCult.Caching.Tests/CultCacheJournalTests.cs
View file

@ -124,12 +124,10 @@ namespace GameCult.Caching.Tests
{
using var cache = Memory();
var calls = 0;
using var journal = cache.AddJournal(changes =>
{
calls++;
Assert.That(changes, Is.Not.Empty);
});
using var journal = cache.AddJournal(_ => calls++);
// Attaching a store with nothing in it is an admission with no change.
cache.AddBackingStore(new SingleFileMessagePackBackingStore(Path.Combine(_directory, "empty.cc")));
Assert.That(cache.Remove(KeyOf("absent")), Is.False);
Assert.That(calls, Is.Zero);

30
tests/GameCult.Caching.Tests/CultCacheObserverTests.cs
View file

@ -2,6 +2,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
using MessagePack;
@ -123,17 +124,36 @@ namespace GameCult.Caching.Tests
}
[Test]
public void AnUnsubscribedObserverStopsReceiving()
public void AnUnsubscribedObserverIsReleasedByTheCache()
{
using var cache = new CultCache(Registry);
var seen = new List<string>();
var subscription = cache.Watch<ObservedPing>().Subscribe(change => seen.Add(change.Document!.Text));
var target = Subscribe(cache, out var subscription);
Send(cache, "one");
CollectGarbage();
Assert.That(target.IsAlive, Is.True, "a live subscription is held by the cache");
subscription.Dispose();
Send(cache, "two");
CollectGarbage();
Assert.That(seen, Is.EqualTo(new[] { "one" }));
Assert.That(target.IsAlive, Is.False, "an unsubscribed observer is no longer held");
GC.KeepAlive(cache);
}
// A disposed R3 subscription ignores OnNext, so a stale entry in the cache is invisible to a receiver; it shows as
// a leak: the cache keeps the subscriber alive.
[MethodImpl(MethodImplOptions.NoInlining)]
private static WeakReference Subscribe(CultCache cache, out IDisposable subscription)
{
var target = new object();
subscription = cache.Watch<ObservedPing>().Subscribe(_ => GC.KeepAlive(target));
return new WeakReference(target);
}
private static void CollectGarbage()
{
GC.Collect();
GC.WaitForPendingFinalizers();
GC.Collect();
}
// A store that admits a load and then fails: the hold's body throws after the changes were minted.