Selkie.git | t/ | 79-store-dispatch-thread-safety.rakutest

---

use Test;
use lib 'lib';

use Selkie::Store;

# dispatch() is documented as safe to call from any thread: worker
# threads (async fx callbacks, app-owned `start {}` pipelines) push
# events concurrently with the main loop's tick/drain. Before the
# queue was lock-protected, the drain's copy-then-clear could wipe a
# push that landed between the two steps — events were silently lost
# under load (a lost 'image-gen/finished' strands the job UI). These
# tests hammer the queue from many threads and assert nothing is
# dropped, duplicated, or reordered within a thread.
#
# Handlers always run on the main thread (inside $s.tick), so the
# counters below are plain variables — no atomics needed.

plan 4;

subtest "concurrent dispatch from many threads loses no events" => {
    plan 2;
    my $s = Selkie::Store.new;

    my Int $handled = 0;
    $s.register-handler('bump', -> $st, %ev {
        $handled++;
        ();
    });

    my constant THREADS    = 8;
    my constant PER-THREAD = 500;
    my constant TOTAL      = THREADS * PER-THREAD;

    # Tick on the main thread while the workers are still pushing —
    # that interleaving is exactly the race the lock has to win. A
    # barrier-then-drain test would never catch the lost-update bug.
    my @workers = (^THREADS).map: -> $t {
        start {
            for ^PER-THREAD -> $i {
                $s.dispatch('bump', thread => $t, seq => $i);
            }
        }
    };
    my $all-pushed = Promise.allof(@workers);

    my $deadline = now + 60;
    while now < $deadline {
        $s.tick;
        last if $all-pushed.status ~~ Kept && $handled == TOTAL;
    }
    await $all-pushed;
    # Final drain — anything pushed after the last tick above.
    $s.tick;

    is @workers.grep({ .status ~~ Kept }).elems, THREADS,
        "all dispatching workers completed";
    is $handled, TOTAL,
        "every one of {TOTAL} cross-thread dispatches was handled";
};

subtest "per-thread dispatch order is preserved" => {
    plan 8;
    my $s = Selkie::Store.new;

    my %seen-seq;    # thread → last seq handled
    my %in-order;    # thread → Bool
    $s.register-handler('ordered', -> $st, %ev {
        my $t   = %ev<thread>;
        my $seq = %ev<seq>;
        %in-order{$t} //= True;
        %in-order{$t} = False
            if %seen-seq{$t}:exists && %seen-seq{$t} >= $seq;
        %seen-seq{$t} = $seq;
        ();
    });

    my constant THREADS    = 8;
    my constant PER-THREAD = 200;

    my @workers = (^THREADS).map: -> $t {
        start {
            for ^PER-THREAD -> $i {
                $s.dispatch('ordered', thread => $t, seq => $i);
            }
        }
    };
    my $all-pushed = Promise.allof(@workers);

    my $deadline = now + 60;
    while now < $deadline {
        $s.tick;
        last if $all-pushed.status ~~ Kept
            && %seen-seq.elems == THREADS
            && %seen-seq.values.all == PER-THREAD - 1;
    }
    await $all-pushed;
    $s.tick;

    for ^THREADS -> $t {
        ok %in-order{$t} && %seen-seq{$t} == PER-THREAD - 1,
            "thread $t: all {PER-THREAD} events handled in dispatch order";
    }
};

subtest "handlers may re-dispatch while the drain holds no lock" => {
    plan 1;
    my $s = Selkie::Store.new;

    # A handler that dispatches directly (not via the dispatch fx)
    # exercises push-during-drain from the main thread. A non-reentrant
    # Lock held across handler execution would deadlock here.
    my @order;
    $s.register-handler('first', -> $st, %ev {
        @order.push('first');
        $st.dispatch('second');
        ();
    });
    $s.register-handler('second', -> $st, %ev {
        @order.push('second');
        ();
    });

    $s.dispatch('first');
    $s.tick;
    is @order.join(','), 'first,second',
        "re-dispatched event processed in the same tick, no deadlock";
};

subtest "async fx on-success dispatch from worker thread is delivered" => {
    plan 1;
    my $s = Selkie::Store.new;

    my $got;
    $s.register-handler('work/start', -> $st, %ev {
        (async => {
            work       => -> { 'payload-from-worker' },
            on-success => 'work/done',
        },);
    });
    $s.register-handler('work/done', -> $st, %ev {
        $got = %ev<result>;
        ();
    });

    $s.dispatch('work/start');
    my $deadline = now + 60;
    while now < $deadline {
        $s.tick;
        last if $got.defined;
    }
    is $got, 'payload-from-worker',
        "worker-thread on-success dispatch arrived through the locked queue";
};