Offset id + persisted

.github/workflows/test.yml

@@ -0,0 +1,29 @@
+name: Run Unit Tests
+
+on:
+  push:
+    branches:
+      - "**"
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up JDK 21
+        uses: actions/setup-java@v4
+        with:
+          java-version: '21'
+          distribution: 'zulu'
+
+      - name: Setup Gradle
+        uses: gradle/gradle-build-action@v3
+
+      - name: Make gradlew executable
+        run: chmod +x ./gradlew
+
+      - name: Run unit tests
+        run: ./gradlew test --stacktrace

src/main/kotlin/no/iktdev/eventi/events/EventDispatcher.kt

@@ -2,6 +2,7 @@ package no.iktdev.eventi.events
 
 import no.iktdev.eventi.models.DeleteEvent
 import no.iktdev.eventi.models.Event
+import no.iktdev.eventi.models.SignalEvent
 import no.iktdev.eventi.stores.EventStore
 import java.util.UUID
 
@@ -11,6 +12,7 @@ open class EventDispatcher(val eventStore: EventStore) {
         val derivedFromIds = events.mapNotNull { it.metadata.derivedFromId }.flatten().toSet()
         val deletedEventIds = events.filterIsInstance<DeleteEvent>().map { it.deletedEventId }
         val candidates = events
+            .filterNot { it is SignalEvent }
             .filter { it.eventId !in derivedFromIds }
             .filter { it.eventId !in deletedEventIds }
 

src/main/kotlin/no/iktdev/eventi/events/EventPollerImplementation.kt

@@ -8,23 +8,30 @@ import no.iktdev.eventi.stores.EventStore
 import java.time.Duration
 import java.time.Instant
 import java.util.UUID
-import kotlin.collections.iterator
 
 abstract class EventPollerImplementation(
     private val eventStore: EventStore,
     private val dispatchQueue: SequenceDispatchQueue,
     private val dispatcher: EventDispatcher
 ) {
-    // Erstatter ikke lastSeenTime, men supplerer den
+    private val log = KotlinLogging.logger {}
-    protected val refWatermark = mutableMapOf<UUID, Instant>()
 
-    // lastSeenTime brukes kun som scan hint
+    /**
+     * Per-reference watermark:
+     *  - first = last seen persistedAt
+     *  - second = last seen persistedId
+     */
+    protected val refWatermark = mutableMapOf<UUID, Pair<Instant, Long>>()
+
+    /**
+     * Global scan hint (timestamp only).
+     * Used to avoid scanning entire table every time.
+     */
     var lastSeenTime: Instant = Instant.EPOCH
 
     open var backoff = Duration.ofSeconds(2)
         protected set
     private val maxBackoff = Duration.ofMinutes(1)
-    private val log = KotlinLogging.logger {}
 
     open suspend fun start() {
         log.info { "EventPoller starting with initial backoff=$backoff" }
@@ -32,7 +39,7 @@ abstract class EventPollerImplementation(
             try {
                 pollOnce()
             } catch (e: Exception) {
-                e.printStackTrace()
+                log.error(e) { "Error in poller loop" }
                 delay(backoff.toMillis())
                 backoff = backoff.multipliedBy(2).coerceAtMost(maxBackoff)
             }
@@ -43,11 +50,11 @@ abstract class EventPollerImplementation(
         val pollStartedAt = MyTime.utcNow()
         log.debug { "🔍 Polling for new events" }
 
-        // Global scan hint: kombiner refWatermark og lastSeenTime
+        // Determine global scan start
-        val watermarkMin = refWatermark.values.minOrNull()
+        val minRefTs = refWatermark.values.minOfOrNull { it.first }
-        val scanFrom = when (watermarkMin) {
+        val scanFrom = when (minRefTs) {
             null -> lastSeenTime
-            else -> maxOf(lastSeenTime, watermarkMin)
+            else -> maxOf(lastSeenTime, minRefTs)
         }
 
         val newPersisted = eventStore.getPersistedEventsAfter(scanFrom)
@@ -59,76 +66,73 @@ abstract class EventPollerImplementation(
             return
         }
 
-        // Vi har sett nye events globalt – reset backoff
+        // Reset backoff
         backoff = Duration.ofSeconds(2)
         log.debug { "📬 Found ${newPersisted.size} new events after $scanFrom" }
 
         val grouped = newPersisted.groupBy { it.referenceId }
         var anyProcessed = false
 
-        // Track høyeste persistedAt vi har sett i denne runden
+        // Track highest persistedAt seen globally this round
         val maxPersistedThisRound = newPersisted.maxOf { it.persistedAt }
 
         for ((ref, eventsForRef) in grouped) {
-            val refSeen = refWatermark[ref] ?: Instant.EPOCH
+            val (refSeenAt, refSeenId) = refWatermark[ref] ?: (Instant.EPOCH to 0L)
+
+            // Filter new events using (timestamp, id) ordering
+            val newForRef = eventsForRef.filter { ev ->
+                ev.persistedAt > refSeenAt ||
+                        (ev.persistedAt == refSeenAt && ev.id > refSeenId)
+            }
 
-            // Finn kun nye events for denne ref’en
-            val newForRef = eventsForRef.filter { it.persistedAt > refSeen }
             if (newForRef.isEmpty()) {
-                log.debug { "🧊 No new events for $ref since $refSeen" }
+                log.debug { "🧊 No new events for $ref since ($refSeenAt, id=$refSeenId)" }
                 continue
             }
 
-            // Hvis ref er busy → ikke oppdater watermark, ikke dispatch
+            // If ref is busy, skip dispatch
             if (dispatchQueue.isProcessing(ref)) {
                 log.debug { "⏳ $ref is busy — deferring ${newForRef.size} events" }
                 continue
             }
 
-            // Hent full sekvens for ref (Eventi-invariant)
+            // Fetch full sequence for dispatch
             val fullLog = eventStore.getPersistedEventsFor(ref)
             val events = fullLog.mapNotNull { it.toEvent() }
 
             log.debug { "🚀 Dispatching ${events.size} events for $ref" }
             dispatchQueue.dispatch(ref, events, dispatcher)
 
-            // Oppdater watermark for denne ref’en
+            // Update watermark for this reference
-            val maxPersistedAtForRef = newForRef.maxOf { it.persistedAt }
+            val maxEvent = newForRef.maxWith(
-            val newWatermark = minOf(pollStartedAt, maxPersistedAtForRef).plusNanos(1)
+                compareBy({ it.persistedAt }, { it.id })
+            )
 
-            refWatermark[ref] = newWatermark
+            val newWatermarkAt = minOf(pollStartedAt, maxEvent.persistedAt)
+            val newWatermarkId = maxEvent.id
+
+            refWatermark[ref] = newWatermarkAt to newWatermarkId
             anyProcessed = true
 
-            log.debug { "⏩ Updated watermark for $ref → $newWatermark" }
+            log.debug { "⏩ Updated watermark for $ref → ($newWatermarkAt, id=$newWatermarkId)" }
         }
 
-        // Oppdater global scan hint uansett – vi har sett nye events
+        // Update global scan hint
-        // Dette hindrer livelock når alle events er <= watermark for sine refs
         val newLastSeen = maxOf(
             lastSeenTime,
             maxPersistedThisRound.plusNanos(1)
         )
 
         if (anyProcessed) {
-            // Behold intensjonen din: globalt hint basert på laveste watermark,
+            val minRef = refWatermark.values.minOfOrNull { it.first }
-            // men aldri gå bakover i tid ift lastSeenTime
+            lastSeenTime = when (minRef) {
-            val minRefWatermark = refWatermark.values.minOrNull()
-            lastSeenTime = when (minRefWatermark) {
                 null -> newLastSeen
-                else -> maxOf(newLastSeen, minRefWatermark)
+                else -> maxOf(newLastSeen, minRef)
             }
             log.debug { "📉 Global scanFrom updated → $lastSeenTime (anyProcessed=true)" }
         } else {
-            // Ingen refs prosessert, men vi vet at alle events vi så er <= watermark
-            // → trygt å flytte lastSeenTime forbi dem
             lastSeenTime = newLastSeen
             log.debug { "🔁 No refs processed — advancing global scanFrom to $lastSeenTime" }
         }
     }
-
-
-
-
-
-
 }

src/main/kotlin/no/iktdev/eventi/models/Event.kt

@@ -4,7 +4,7 @@ import java.util.UUID
 
 @Suppress("UNCHECKED_CAST")
 abstract class Event {
-    var referenceId: UUID = UUID.randomUUID()
+    lateinit var referenceId: UUID
         protected set
     var eventId: UUID = UUID.randomUUID()
         private set
@@ -43,6 +43,6 @@ abstract class DeleteEvent(
     open val deletedEventId: UUID
 ) : Event()
 
-
+abstract class SignalEvent(): Event()
 
 

src/main/kotlin/no/iktdev/eventi/tasks/TaskListener.kt

@@ -9,6 +9,7 @@ import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import no.iktdev.eventi.models.Event
 import no.iktdev.eventi.models.Task
+import no.iktdev.eventi.models.store.TaskStatus
 import org.jetbrains.annotations.VisibleForTesting
 import java.util.UUID
 import kotlin.coroutines.cancellation.CancellationException
@@ -67,9 +68,7 @@ abstract class TaskListener(val taskType: TaskType = TaskType.CPU_INTENSIVE): Ta
         currentJob = getDispatcherForTask(task).launch {
             try {
                 val result = onTask(task)
-                reporter.markCompleted(task.taskId)
                 onComplete(task, result)
-
             } catch (e: CancellationException) {
                 // Dette er en ekte kansellering
                 onCancelled(task)
@@ -88,14 +87,16 @@ abstract class TaskListener(val taskType: TaskType = TaskType.CPU_INTENSIVE): Ta
                 this@TaskListener.reporter = null
             }
         }
-
         return true
     }
 
+    abstract fun createIncompleteStateTaskEvent(task: Task, status: TaskStatus, exception: Exception? = null): Event
+
     override fun onError(task: Task, exception: Exception) {
         reporter?.log(task.taskId, "Error processing task: ${exception.message}")
         exception.printStackTrace()
-        reporter?.markFailed(task.taskId)
+        reporter?.markFailed(task.referenceId, task.taskId)
+        reporter!!.publishEvent(createIncompleteStateTaskEvent(task, TaskStatus.Failed, exception))
     }
 
     override fun onComplete(task: Task, result: Event?) {
@@ -107,10 +108,11 @@ abstract class TaskListener(val taskType: TaskType = TaskType.CPU_INTENSIVE): Ta
     }
 
     override fun onCancelled(task: Task) {
-        reporter!!.markCancelled(task.taskId)
+        reporter!!.markCancelled(task.referenceId, task.taskId)
         currentJob?.cancel()
         heartbeatRunner?.cancel()
         currentTask = null
+        reporter!!.publishEvent(createIncompleteStateTaskEvent(task, TaskStatus.Cancelled))
     }
 }
 
@@ -134,8 +136,8 @@ interface TaskReporter {
     fun markClaimed(taskId: UUID, workerId: String)
     fun updateLastSeen(taskId: UUID)
     fun markCompleted(taskId: UUID)
-    fun markFailed(taskId: UUID)
+    fun markFailed(referenceId: UUID, taskId: UUID)
-    fun markCancelled(taskId: UUID)
+    fun markCancelled(referenceId: UUID, taskId: UUID)
     fun updateProgress(taskId: UUID, progress: Int)
     fun log(taskId: UUID, message: String)
     fun publishEvent(event: Event)

src/test/kotlin/no/iktdev/eventi/EventDispatcherTest.kt

@@ -8,7 +8,7 @@ import no.iktdev.eventi.events.EventListenerRegistry
 import no.iktdev.eventi.events.EventTypeRegistry
 import no.iktdev.eventi.models.DeleteEvent
 import no.iktdev.eventi.models.Event
-import no.iktdev.eventi.models.Metadata
+import no.iktdev.eventi.models.SignalEvent
 import no.iktdev.eventi.testUtil.wipe
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -20,12 +20,14 @@ import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 import java.util.UUID
 
-@DisplayName("""
+@DisplayName(
+    """
 EventDispatcher
 Når hendelser dispatches til lyttere
 Hvis hendelsene inneholder avledede, slettede eller nye events
 Så skal dispatcheren håndtere filtrering, replays og historikk korrekt
-""")
+"""
+)
 class EventDispatcherTest : TestBase() {
 
     val dispatcher = EventDispatcher(eventStore)
@@ -34,9 +36,6 @@ class EventDispatcherTest : TestBase() {
     class TriggerEvent : Event()
     class OtherEvent : Event()
     class DummyEvent : Event() {
-        fun putMetadata(metadata: Metadata) {
-            this.metadata = metadata
-        }
     }
 
     @BeforeEach
@@ -55,15 +54,17 @@ class EventDispatcherTest : TestBase() {
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en TriggerEvent dispatches
     Hvis en lytter produserer én DerivedEvent
     Så skal kun én ny event produseres og prosessen stoppe
-    """)
+    """
+    )
     fun shouldProduceOneEventAndStop() {
-        val listener = ProducingListener()
+        ProducingListener()
 
-        val trigger = TriggerEvent()
+        val trigger = TriggerEvent().newReferenceId()
         dispatcher.dispatch(trigger.referenceId, listOf(trigger))
 
         val produced = eventStore.all().firstOrNull()
@@ -71,55 +72,63 @@ class EventDispatcherTest : TestBase() {
 
         val event = produced!!.toEvent()
         assertThat(event!!.metadata.derivedFromId).hasSize(1)
-        assertThat(event!!.metadata.derivedFromId).contains(trigger.eventId)
+        assertThat(event.metadata.derivedFromId).contains(trigger.eventId)
         assertTrue(event is DerivedEvent)
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en event allerede har avledet en DerivedEvent
     Hvis dispatcheren replays historikken
     Så skal ikke DerivedEvent produseres på nytt
-    """)
+    """
+    )
     fun shouldSkipAlreadyDerivedEvents() {
-        val listener = ProducingListener()
+        ProducingListener()
 
-        val trigger = TriggerEvent()
+        val trigger = TriggerEvent().newReferenceId()
         val derived = DerivedEvent().derivedOf(trigger).toPersisted(1L, MyTime.utcNow())
 
         eventStore.persist(derived!!.toEvent()!!) // simulate prior production
 
-        dispatcher.dispatch(trigger.referenceId, listOf(trigger, derived!!.toEvent()!!))
+        dispatcher.dispatch(trigger.referenceId, listOf(trigger, derived.toEvent()!!))
 
         assertEquals(1, eventStore.all().size)
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når flere events dispatches
     Hvis en lytter mottar en event
     Så skal hele historikken leveres i context
-    """)
+    """
+    )
     fun shouldPassFullContextToListener() {
         val listener = ContextCapturingListener()
 
-        val e1 = TriggerEvent()
+        val e1 = TriggerEvent().newReferenceId()
-        val e2 = OtherEvent()
+        val e2 = OtherEvent().newReferenceId()
         dispatcher.dispatch(e1.referenceId, listOf(e1, e2))
 
         assertEquals(2, listener.context.size)
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en replay skjer
     Hvis en event allerede har produsert en DerivedEvent
     Så skal ikke DerivedEvent produseres på nytt
-    """)
+    """
+    )
     fun shouldBehaveDeterministicallyAcrossReplays() {
-        val listener = ProducingListener()
+        val referenceId = UUID.randomUUID()
 
-        val trigger = TriggerEvent()
+        ProducingListener()
+
+        val trigger = TriggerEvent().usingReferenceId(referenceId)
         dispatcher.dispatch(trigger.referenceId, listOf(trigger))
         val replayContext = listOf(trigger) + eventStore.all().mapNotNull { it.toEvent() }
 
@@ -129,12 +138,16 @@ class EventDispatcherTest : TestBase() {
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en DeleteEvent peker på en tidligere event
     Hvis dispatcheren filtrerer kandidater
     Så skal slettede events ikke leveres som kandidater
-    """)
+    """
+    )
     fun shouldNotDeliverDeletedEventsAsCandidates() {
+        val referenceId = UUID.randomUUID()
+
         val dispatcher = EventDispatcher(eventStore)
         val received = mutableListOf<Event>()
 
@@ -145,11 +158,10 @@ class EventDispatcherTest : TestBase() {
             }
         }
         // Original hendelse
-        val original = TriggerEvent()
+        val original = TriggerEvent().usingReferenceId(referenceId)
 
         // Slettehendelse som peker på original
-        val deleted = object : DeleteEvent(original.eventId) {
+        val deleted = object : DeleteEvent(original.eventId) {}.apply { newReferenceId() }
-        }
 
         // Dispatch med begge hendelser
         dispatcher.dispatch(original.referenceId, listOf(original, deleted))
@@ -166,36 +178,42 @@ class EventDispatcherTest : TestBase() {
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en DeleteEvent dispatches alene
     Hvis en lytter reagerer på DeleteEvent
     Så skal DeleteEvent leveres som kandidat
-    """)
+    """
+    )
     fun shouldDeliverDeleteEventToListenersThatReactToIt() {
         val received = mutableListOf<Event>()
-        val listener = object : EventListener() {
+        val referenceId = UUID.randomUUID()
-            override fun onEvent(event: Event, context: List<Event>): Event? {
+
+        object : EventListener() {
+            override fun onEvent(event: Event, history: List<Event>): Event? {
                 if (event is DeleteEvent) received += event
                 return null
             }
         }
 
-        val deleted = object : DeleteEvent(UUID.randomUUID()) {}
+        val deleted = object : DeleteEvent(UUID.randomUUID()) {}.apply { usingReferenceId(referenceId) }
         dispatcher.dispatch(deleted.referenceId, listOf(deleted))
 
         assertTrue(received.contains(deleted))
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en event har avledet en ny event
     Hvis dispatcheren replays historikken
     Så skal ikke original-eventen leveres som kandidat igjen
-    """)
+    """
+    )
     fun shouldNotRedeliverEventsThatHaveProducedDerivedEvents() {
-        val listener = ProducingListener()
+        ProducingListener()
 
-        val trigger = TriggerEvent()
+        val trigger = TriggerEvent().newReferenceId()
         // Første dispatch: trigger produserer en DerivedEvent
         dispatcher.dispatch(trigger.referenceId, listOf(trigger))
 
@@ -215,20 +233,22 @@ class EventDispatcherTest : TestBase() {
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en DeleteEvent slettet en tidligere event
     Hvis dispatcheren bygger historikk
     Så skal slettede events ikke være med i history
-    """)
+    """
+    )
     fun historyShouldExcludeDeletedEvents() {
         val dispatcher = EventDispatcher(eventStore)
 
-        val original = TriggerEvent()
+        val original = TriggerEvent().newReferenceId()
-        val deleted = object : DeleteEvent(original.eventId) {}
+        val deleted = object : DeleteEvent(original.eventId) {}.apply { usingReferenceId(original.referenceId) }
 
         var receivedHistory: List<Event> = emptyList()
 
-        val listener = object : EventListener() {
+        object : EventListener() {
             override fun onEvent(event: Event, history: List<Event>): Event? {
                 receivedHistory = history
                 return null
@@ -242,21 +262,23 @@ class EventDispatcherTest : TestBase() {
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en DeleteEvent slettet en event
     Hvis andre events fortsatt er gyldige
     Så skal history kun inneholde de ikke-slettede events
-    """)
+    """
+    )
     fun historyShouldKeepNonDeletedEvents() {
         val dispatcher = EventDispatcher(eventStore)
-
+        val referenceId = UUID.randomUUID()
-        val e1 = TriggerEvent()
+        val e1 = TriggerEvent().usingReferenceId(referenceId)
-        val e2 = OtherEvent()
+        val e2 = OtherEvent().usingReferenceId(referenceId)
         val deleted = object : DeleteEvent(e1.eventId) {}
 
         var receivedHistory: List<Event> = emptyList()
 
-        val listener = object : EventListener() {
+        object : EventListener() {
             override fun onEvent(event: Event, history: List<Event>): Event? {
                 receivedHistory = history
                 return null
@@ -271,21 +293,23 @@ class EventDispatcherTest : TestBase() {
     }
 
     @Test
-    @DisplayName("""
+    @DisplayName(
+        """
     Når en DeleteEvent er kandidat
     Hvis historikken kun inneholder slettede events
     Så skal history være tom
-    """)
+    """
+    )
     fun deleteEventShouldBeDeliveredButHistoryEmpty() {
         val dispatcher = EventDispatcher(eventStore)
 
-        val original = TriggerEvent()
+        val original = TriggerEvent().newReferenceId()
-        val deleted = object : DeleteEvent(original.eventId) {}
+        val deleted = object : DeleteEvent(original.eventId) {}.apply { newReferenceId() }
 
         var receivedEvent: Event? = null
         var receivedHistory: List<Event> = emptyList()
 
-        val listener = object : EventListener() {
+        object : EventListener() {
             override fun onEvent(event: Event, history: List<Event>): Event? {
                 receivedEvent = event
                 receivedHistory = history
@@ -299,18 +323,66 @@ class EventDispatcherTest : TestBase() {
         assertTrue(receivedHistory.isEmpty())
     }
 
+    @Test
+    @DisplayName(
+    """
+    Når en SignalEvent dispatches
+    Hvis SignalEvent ikke skal være kandidat
+    Så skal den ikke leveres til lyttere, men fortsatt være i historikken
+    """
+    )
+    fun shouldNotDeliverSignalEventAsCandidate() {
+        // Arrange
+        class TestSignalEvent : SignalEvent()
+        EventTypeRegistry.register(listOf(TestSignalEvent::class.java,))
+
+        val received = mutableListOf<Event>()
+        var finalHistory: List<Event>? = null
+        object : EventListener() {
+            override fun onEvent(event: Event, history: List<Event>): Event? {
+                received += event
+                finalHistory = history
+                return null
+            }
+        }
+
+        val refId = UUID.randomUUID()
+        val trigger = TriggerEvent().usingReferenceId(refId)
+        val signal = TestSignalEvent().usingReferenceId(refId)
+
+        // Act
+        dispatcher.dispatch(trigger.referenceId, listOf(trigger, signal))
+
+        // Assert
+        // 1) TriggerEvent skal leveres
+        assertTrue(received.any { it is TriggerEvent }) {
+            "TriggerEvent skal leveres som kandidat"
+        }
+
+        // 2) SignalEvent skal IKKE leveres
+        assertFalse(received.any { it is TestSignalEvent }) {
+            "SignalEvent skal ikke leveres som kandidat"
+        }
+
+        assertNotNull(finalHistory)
+        assertTrue(finalHistory!!.any { it is TestSignalEvent }) {
+            "SignalEvent skal være i historikken selv om den ikke er kandidat"
+        }
+    }
+
+
     // --- Test helpers ---
 
     class ProducingListener : EventListener() {
-        override fun onEvent(event: Event, context: List<Event>): Event? {
+        override fun onEvent(event: Event, history: List<Event>): Event? {
             return if (event is TriggerEvent) DerivedEvent().derivedOf(event) else null
         }
     }
 
     class ContextCapturingListener : EventListener() {
         var context: List<Event> = emptyList()
-        override fun onEvent(event: Event, context: List<Event>): Event? {
+        override fun onEvent(event: Event, history: List<Event>): Event? {
-            this.context = context
+            this.context = history
             return null
         }
     }

src/test/kotlin/no/iktdev/eventi/ZDSTest.kt

@@ -39,7 +39,7 @@ class ZDSTest {
     fun scenario1() {
         EventTypeRegistry.register(EchoEvent::class.java)
 
-        val echo = EchoEvent("hello")
+        val echo = EchoEvent("hello").newReferenceId()
         val persisted = echo.toPersisted(id = 1L)
 
         val restored = persisted!!.toEvent()

src/test/kotlin/no/iktdev/eventi/events/EventPollerImplementationTest.kt

@@ -1,20 +1,23 @@
 package no.iktdev.eventi.events
 
 import kotlinx.coroutines.CompletableDeferred
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
-import no.iktdev.eventi.EventDispatcherTest
+import kotlinx.coroutines.awaitAll
 import no.iktdev.eventi.EventDispatcherTest.DerivedEvent
 import no.iktdev.eventi.EventDispatcherTest.OtherEvent
 import no.iktdev.eventi.EventDispatcherTest.TriggerEvent
 import no.iktdev.eventi.MyTime
 import no.iktdev.eventi.TestBase
 import no.iktdev.eventi.models.Event
+import no.iktdev.eventi.testUtil.TestSequenceDispatchQueue
 import no.iktdev.eventi.testUtil.wipe
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
@@ -34,9 +37,7 @@ Så skal polleren dispatch'e riktig, oppdatere lastSeenTime og unngå duplikater
 """)
 class EventPollerImplementationTest : TestBase() {
 
-    val dispatcher = EventDispatcher(eventStore)
+    private val dispatcher = EventDispatcher(eventStore)
-    val queue = SequenceDispatchQueue(maxConcurrency = 8)
-    val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
 
     @BeforeEach
     fun setup() {
@@ -60,12 +61,16 @@ class EventPollerImplementationTest : TestBase() {
     Så skal alle referenceId-er dispatch'es og lastSeenTime oppdateres
     """)
     fun pollOnceDispatchesAllNewReferenceIdsAndUpdatesLastSeenTime() = runTest {
+        val testDispatcher = StandardTestDispatcher(testScheduler)
+        val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
+        val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
+
         val dispatched = ConcurrentHashMap.newKeySet<UUID>()
         val completionMap = mutableMapOf<UUID, CompletableDeferred<Unit>>()
 
         EventListenerRegistry.registerListener(
             object : EventListener() {
-                override fun onEvent(event: Event, context: List<Event>): Event? {
+                override fun onEvent(event: Event, history: List<Event>): Event? {
                     dispatched += event.referenceId
                     completionMap[event.referenceId]?.complete(Unit)
                     return null
@@ -94,6 +99,9 @@ class EventPollerImplementationTest : TestBase() {
     Så skal backoff øke, og resettes når nye events ankommer
     """)
     fun pollOnceIncreasesBackoffWhenNoEventsAndResetsWhenEventsArrive() = runTest {
+        val testDispatcher = StandardTestDispatcher(testScheduler)
+        val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
+
         val testPoller = object : EventPollerImplementation(eventStore, queue, dispatcher) {
             fun currentBackoff(): Duration = backoff
         }
@@ -122,6 +130,10 @@ class EventPollerImplementationTest : TestBase() {
     Så skal polleren gruppere og dispatch'e alle tre i én batch
     """)
     fun pollOnceGroupsAndDispatchesExactlyThreeEventsForOneReferenceId() = runTest {
+        val testDispatcher = StandardTestDispatcher(testScheduler)
+        val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
+        val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
+
         val refId = UUID.randomUUID()
         val received = mutableListOf<Event>()
         val done = CompletableDeferred<Unit>()
@@ -133,7 +145,7 @@ class EventPollerImplementationTest : TestBase() {
         EventTypeRegistry.register(listOf(TriggerEvent::class.java))
 
         object : EventListener() {
-            override fun onEvent(event: Event, context: List<Event>): Event? {
+            override fun onEvent(event: Event, history: List<Event>): Event? {
                 received += event
                 if (received.size == 3) done.complete(Unit)
                 return null
@@ -158,8 +170,8 @@ class EventPollerImplementationTest : TestBase() {
     Så skal polleren ignorere dem
     """)
     fun pollOnceIgnoresEventsBeforeLastSeenTime() = runTest {
-        val refId = UUID.randomUUID()
+        val testDispatcher = StandardTestDispatcher(testScheduler)
-        val ignored = TriggerEvent().usingReferenceId(refId)
+        val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
 
         val testPoller = object : EventPollerImplementation(eventStore, queue, dispatcher) {
             init {
@@ -167,6 +179,9 @@ class EventPollerImplementationTest : TestBase() {
             }
         }
 
+        val refId = UUID.randomUUID()
+        val ignored = TriggerEvent().usingReferenceId(refId)
+
         eventStore.persist(ignored)
         testPoller.pollOnce()
 
@@ -181,12 +196,17 @@ class EventPollerImplementationTest : TestBase() {
     Så skal begge events prosesseres, men uten å produsere duplikate derived events
     """)
     fun pollerHandlesManuallyInjectedDuplicateEvent() = runTest {
+        val testDispatcher = StandardTestDispatcher(testScheduler)
+        val queue = TestSequenceDispatchQueue(maxConcurrency = 8, dispatcher = testDispatcher)
+        val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
+
         EventTypeRegistry.register(listOf(MarcoEvent::class.java, EchoEvent::class.java))
+
         val channel = Channel<Event>(Channel.UNLIMITED)
         val handled = mutableListOf<Event>()
 
         object : EventListener() {
-            override fun onEvent(event: Event, context: List<Event>): Event? {
+            override fun onEvent(event: Event, history: List<Event>): Event? {
                 if (event !is EchoEvent) return null
                 handled += event
                 channel.trySend(event)
@@ -194,16 +214,14 @@ class EventPollerImplementationTest : TestBase() {
             }
         }
 
-        val poller = object : EventPollerImplementation(eventStore, queue, dispatcher) {}
+        val original = EchoEvent("Hello").newReferenceId()
-
-        val original = EchoEvent("Hello")
         eventStore.persist(original)
 
         poller.pollOnce()
 
-        withContext(Dispatchers.Default.limitedParallelism(1)) {
+        withContext(testDispatcher) {
-            withTimeout(Duration.ofMinutes(1).toMillis()) {
+            withTimeout(60_000) {
-                repeat(1) { channel.receive() }
+                channel.receive()
             }
         }
 
@@ -212,9 +230,9 @@ class EventPollerImplementationTest : TestBase() {
 
         poller.pollOnce()
 
-        withContext(Dispatchers.Default.limitedParallelism(1)) {
+        withContext(testDispatcher) {
-            withTimeout(Duration.ofMinutes(1).toMillis()) {
+            withTimeout(60_000) {
-                repeat(1) { channel.receive() }
+                channel.receive()
             }
         }
 

src/test/kotlin/no/iktdev/eventi/events/RunSimulationTest.kt

@@ -1,7 +1,7 @@
+@file:OptIn(ExperimentalCoroutinesApi::class)
+
 package no.iktdev.eventi.events
 
-import io.mockk.every
-import io.mockk.mockk
 import kotlinx.coroutines.*
 import kotlinx.coroutines.test.*
 import no.iktdev.eventi.InMemoryEventStore
@@ -9,9 +9,6 @@ import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import java.util.UUID
-import kotlinx.coroutines.*
-import no.iktdev.eventi.MyTime
-import no.iktdev.eventi.ZDS.toPersisted
 import no.iktdev.eventi.models.Event
 import no.iktdev.eventi.models.Metadata
 import org.junit.jupiter.api.DisplayName
@@ -89,7 +86,7 @@ class RunSimulationTestTest {
         }
     }
 
-    private fun persistEvent(ref: UUID, time: Instant) {
+    private fun persistEvent(ref: UUID) {
         val e = TestEvent().withReference(ref)
         store.persist(e.setMetadata(Metadata()))
     }
@@ -102,9 +99,8 @@ class RunSimulationTestTest {
     """)
     fun pollerUpdatesLastSeenTimeWhenDispatchHappens() = runTest(testDispatcher) {
         val ref = UUID.randomUUID()
-        val t = Instant.parse("2026-01-22T12:00:00Z")
 
-        persistEvent(ref, t)
+        persistEvent(ref)
 
         poller.pollOnce()
         advanceUntilIdle()
@@ -149,9 +145,8 @@ class RunSimulationTestTest {
     """)
     fun pollerDoesNotDoubleDispatch() = runTest(testDispatcher) {
         val ref = UUID.randomUUID()
-        val t = Instant.parse("2026-01-22T12:00:00Z")
 
-        persistEvent(ref, t)
+        persistEvent(ref)
 
         poller.pollOnce()
         advanceUntilIdle()
@@ -171,10 +166,9 @@ class RunSimulationTestTest {
     fun pollerHandlesMultipleReferenceIds() = runTest(testDispatcher) {
         val refA = UUID.randomUUID()
         val refB = UUID.randomUUID()
-        val t = Instant.parse("2026-01-22T12:00:00Z")
 
-        persistEvent(refA, t)
+        persistEvent(refA)
-        persistEvent(refB, t.plusSeconds(1))
+        persistEvent(refB)
 
         poller.pollOnce()
         advanceUntilIdle()
@@ -191,10 +185,9 @@ class RunSimulationTestTest {
     fun pollerHandlesIdenticalTimestamps() = runTest(testDispatcher) {
         val refA = UUID.randomUUID()
         val refB = UUID.randomUUID()
-        val t = Instant.parse("2026-01-22T12:00:00Z")
 
-        persistEvent(refA, t)
+        persistEvent(refA)
-        persistEvent(refB, t)
+        persistEvent(refB)
 
         poller.pollOnce()
         advanceUntilIdle()
@@ -241,10 +234,8 @@ class RunSimulationTestTest {
     """)
     fun pollerProcessesEventsArrivingWhileQueueBusy() = runTest(testDispatcher) {
         val ref = UUID.randomUUID()
-        val t1 = Instant.parse("2026-01-22T12:00:00Z")
-        val t2 = t1.plusSeconds(5)
 
-        persistEvent(ref, t1)
+        persistEvent(ref)
 
         val controlledQueue = ControlledDispatchQueue(scope)
         controlledQueue.busyRefs += ref
@@ -261,7 +252,7 @@ class RunSimulationTestTest {
         controlledQueue.busyRefs.clear()
 
         // Add new event
-        persistEvent(ref, t2)
+        persistEvent(ref)
 
         // Poll #2: should dispatch both events
         poller.pollOnce()

src/test/kotlin/no/iktdev/eventi/events/SequenceDispatchQueueTest.kt

@@ -51,7 +51,7 @@ class SequenceDispatchQueueTest : TestBase() {
 
         EventListenerRegistry.registerListener(
             object : EventListener() {
-                override fun onEvent(event: Event, context: List<Event>): Event? {
+                override fun onEvent(event: Event, history: List<Event>): Event? {
                     dispatched += event.referenceId
                     Thread.sleep(50) // simuler tung prosessering
                     return null

src/test/kotlin/no/iktdev/eventi/events/poller/PollerStartLoopTest.kt

@@ -1,5 +1,6 @@
 package no.iktdev.eventi.events.poller
 
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.*
 import no.iktdev.eventi.InMemoryEventStore
 import no.iktdev.eventi.MyTime
@@ -23,6 +24,7 @@ import org.assertj.core.api.Assertions.assertThat
 import java.time.Duration
 
 
+@ExperimentalCoroutinesApi
 @DisplayName("""
 EventPollerImplementation – start-loop
 Når polleren kjører i en kontrollert test-loop
@@ -58,6 +60,42 @@ class PollerStartLoopTest : TestBase() {
         store.persistAt(e, time)
     }
 
+
+    @Test
+    @DisplayName("""
+    Når to events har identisk persistedAt
+    Hvis polleren kjører
+    Så skal begge events prosesseres og ingen mistes
+    """)
+    fun `poller handles same-timestamp events without losing any`() = runTest {
+        val ref = UUID.randomUUID()
+        val ts = Instant.parse("2025-01-01T12:00:00Z")
+
+        // Two events with same timestamp but different IDs
+        val e1 = TestEvent().withReference(ref).setMetadata(Metadata())
+        val e2 = TestEvent().withReference(ref).setMetadata(Metadata())
+
+        store.persistAt(e1, ts) // id=1
+        store.persistAt(e2, ts) // id=2
+
+        poller.startFor(iterations = 1)
+
+        // Verify dispatch happened
+        assertThat(dispatcher.dispatched).hasSize(1)
+
+        val (_, events) = dispatcher.dispatched.single()
+
+        // Both events must be present
+        assertThat(events.map { it.eventId })
+            .hasSize(2)
+            .doesNotHaveDuplicates()
+
+        // Watermark must reflect highest ID
+        val wm = poller.watermarkFor(ref)
+        assertThat(wm!!.first).isEqualTo(ts)
+        assertThat(wm.second).isEqualTo(2)
+    }
+
     @Test
     @DisplayName("""
     Når polleren kjører flere iterasjoner uten events
@@ -100,7 +138,6 @@ class PollerStartLoopTest : TestBase() {
     """)
     fun `poller resets backoff when events appear`() = runTest {
         poller.startFor(iterations = 5)
-        val before = poller.backoff
 
         val ref = UUID.randomUUID()
         persistAt(ref, MyTime.utcNow())
@@ -270,11 +307,15 @@ class PollerStartLoopTest : TestBase() {
 
         poller.startFor(iterations = 1)
 
+
+
         // A skal IKKE ha flyttet watermark
         assertThat(poller.watermarkFor(refA)).isEqualTo(wmA1)
 
-        // B skal ha flyttet watermark
+        // B skal ha flyttet watermark (på timestamp-nivå)
-        assertThat(poller.watermarkFor(refB)).isGreaterThan(wmB1)
+        val wmB2 = poller.watermarkFor(refB)
+        assertThat(wmB2!!.first).isGreaterThan(wmB1!!.first)
+
     }
 
     @DisplayName("🍌 Bananastesten™ — stress-test av watermark, busy refs og dispatch-semantikk")
@@ -404,7 +445,7 @@ class PollerStartLoopTest : TestBase() {
 
         // Fake EventStore som alltid returnerer samme event
         val fakeStore = object : EventStore {
-            override fun getPersistedEventsAfter(ts: Instant): List<PersistedEvent> {
+            override fun getPersistedEventsAfter(timestamp: Instant): List<PersistedEvent> {
                 // Alltid returner én event som ligger før watermark
                 return listOf(
                     PersistedEvent(
@@ -418,7 +459,7 @@ class PollerStartLoopTest : TestBase() {
                 )
             }
 
-            override fun getPersistedEventsFor(ref: UUID): List<PersistedEvent> = emptyList()
+            override fun getPersistedEventsFor(referenceId: UUID): List<PersistedEvent> = emptyList()
             override fun persist(event: Event) = Unit
         }
 
@@ -432,7 +473,7 @@ class PollerStartLoopTest : TestBase() {
 
         // Sett watermark høyt (polleren setter watermark selv i ekte drift,
         // men i denne testen må vi simulere det)
-        poller.setWatermarkFor(ref, t(100))
+        poller.setWatermarkFor(ref, t(100), id = 999)
 
         // Sett lastSeenTime bak eventen
         poller.lastSeenTime = t(0)

src/test/kotlin/no/iktdev/eventi/events/poller/TestablePoller.kt

@@ -1,14 +1,15 @@
 package no.iktdev.eventi.events.poller
 
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.TestScope
-import kotlinx.coroutines.test.advanceTimeBy
 import no.iktdev.eventi.events.EventDispatcher
 import no.iktdev.eventi.events.EventPollerImplementation
 import no.iktdev.eventi.events.SequenceDispatchQueue
 import no.iktdev.eventi.stores.EventStore
 import java.time.Instant
-import java.util.UUID
+import java.util.*
 
+@ExperimentalCoroutinesApi
 class TestablePoller(
     eventStore: EventStore,
     dispatchQueue: SequenceDispatchQueue,
@@ -16,8 +17,6 @@ class TestablePoller(
     val scope: TestScope
 ) : EventPollerImplementation(eventStore, dispatchQueue, dispatcher), WatermarkDebugView {
 
-
-
     suspend fun startFor(iterations: Int) {
         repeat(iterations) {
             try {
@@ -31,19 +30,17 @@ class TestablePoller(
         }
     }
 
-    override fun watermarkFor(ref: UUID): Instant? {
+    override fun watermarkFor(ref: UUID): Pair<Instant, Long>? {
-        return refWatermark[ref]?.let {
+        return refWatermark[ref]
-            return it
+    }
+
+    override fun setWatermarkFor(ref: UUID, time: Instant, id: Long) {
+        refWatermark[ref] = time to id
     }
 }
 
-    override fun setWatermarkFor(ref: UUID, time: Instant) {
-        refWatermark[ref] = time
-    }
-
-
-}
 interface WatermarkDebugView {
-    fun watermarkFor(ref: UUID): Instant?
+    fun watermarkFor(ref: UUID): Pair<Instant, Long>?
-    fun setWatermarkFor(ref: UUID, time: Instant)
+    fun setWatermarkFor(ref: UUID, time: Instant, id: Long)
 }
+

src/test/kotlin/no/iktdev/eventi/tasks/TaskListenerRegistryTest.kt

@@ -5,6 +5,7 @@ import no.iktdev.eventi.events.EventListener
 import no.iktdev.eventi.events.EventListenerRegistry
 import no.iktdev.eventi.models.Event
 import no.iktdev.eventi.models.Task
+import no.iktdev.eventi.models.store.TaskStatus
 import no.iktdev.eventi.testUtil.wipe
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Assertions.*
@@ -19,6 +20,15 @@ class TaskListenerRegistryTest {
         override fun getWorkerId(): String {
             TODO("Not yet implemented")
         }
+
+        override fun createIncompleteStateTaskEvent(
+            task: Task,
+            status: TaskStatus,
+            exception: Exception?
+        ): Event {
+            TODO("Not yet implemented")
+        }
+
         override fun supports(task: Task): Boolean {
             TODO("Not yet implemented")
         }
@@ -32,6 +42,15 @@ class TaskListenerRegistryTest {
         override fun getWorkerId(): String {
             TODO("Not yet implemented")
         }
+
+        override fun createIncompleteStateTaskEvent(
+            task: Task,
+            status: TaskStatus,
+            exception: Exception?
+        ): Event {
+            TODO("Not yet implemented")
+        }
+
         override fun supports(task: Task): Boolean {
             TODO("Not yet implemented")
         }
@@ -45,6 +64,15 @@ class TaskListenerRegistryTest {
         override fun getWorkerId(): String {
             TODO("Not yet implemented")
         }
+
+        override fun createIncompleteStateTaskEvent(
+            task: Task,
+            status: TaskStatus,
+            exception: Exception?
+        ): Event {
+            TODO("Not yet implemented")
+        }
+
         override fun supports(task: Task): Boolean {
             TODO("Not yet implemented")
         }
@@ -57,6 +85,15 @@ class TaskListenerRegistryTest {
         override fun getWorkerId(): String {
             TODO("Not yet implemented")
         }
+
+        override fun createIncompleteStateTaskEvent(
+            task: Task,
+            status: TaskStatus,
+            exception: Exception?
+        ): Event {
+            TODO("Not yet implemented")
+        }
+
         override fun supports(task: Task): Boolean {
             TODO("Not yet implemented")
         }

src/test/kotlin/no/iktdev/eventi/tasks/TaskListenerTest.kt

@@ -2,13 +2,13 @@ package no.iktdev.eventi.tasks
 
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.yield
 import no.iktdev.eventi.models.Event
 import no.iktdev.eventi.models.Task
+import no.iktdev.eventi.models.store.TaskStatus
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
@@ -23,32 +23,29 @@ Så skal state, heartbeat og cleanup fungere korrekt
 """)
 class TaskListenerTest {
 
-    // -------------------------
-    // Fake Task + Reporter
-    // -------------------------
-
     class FakeTask : Task()
 
     class FakeReporter : TaskReporter {
         var claimed = false
-        var consumed = false
+        var completed = false
-        var logs = mutableListOf<String>()
+        var failed = false
-        var events = mutableListOf<Event>()
+        var cancelled = false
+        val logs = mutableListOf<String>()
+        val events = mutableListOf<Event>()
 
         override fun markClaimed(taskId: UUID, workerId: String) { claimed = true }
-        override fun markCompleted(taskId: UUID) { consumed = true }
+        override fun markCompleted(taskId: UUID) { completed = true }
-        override fun markFailed(taskId: UUID) { consumed = true }
+        override fun markFailed(referenceId: UUID, taskId: UUID) { failed = true }
-        override fun markCancelled(taskId: UUID) {}
+        override fun markCancelled(referenceId: UUID, taskId: UUID) { cancelled = true }
         override fun updateProgress(taskId: UUID, progress: Int) {}
         override fun publishEvent(event: Event) { events.add(event) }
         override fun updateLastSeen(taskId: UUID) {}
         override fun log(taskId: UUID, message: String) { logs.add(message) }
     }
 
-    // -------------------------
+    // ---------------------------------------------------------
-    // Tests
+    // 1 — Heartbeat starter og stopper riktig
-    // -------------------------
+    // ---------------------------------------------------------
-
     @Test
     @DisplayName("""
     Når onTask starter heartbeat-runner
@@ -58,13 +55,15 @@ class TaskListenerTest {
     fun heartbeatStartsAndStopsCorrectly() = runTest {
         val listener = object : TaskListener() {
 
-            var heartbeatStarted: Job? = null
+            var heartbeatRan = false
-            var heartbeatRan: Boolean = false
-                private set
-
             var onTaskCalled = false
 
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
             override suspend fun onTask(task: Task): Event? {
@@ -72,25 +71,18 @@ class TaskListenerTest {
 
                 withHeartbeatRunner(10.milliseconds) {
                     heartbeatRan = true
-                }.also { heartbeatStarted = it }
+                }
 
-                // Gi heartbeat en sjanse til å kjøre
                 yield()
-
                 return object : Event() {}
             }
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
+        listener.accept(FakeTask(), reporter)
 
-        val accepted = listener.accept(task, reporter)
+        listener.currentJob?.join()
-        assertTrue(accepted)
 
-        listener.currentJob!!.join()
-
-        assertNotNull(listener.heartbeatStarted)
-        assertFalse(listener.heartbeatStarted!!.isActive)
         assertTrue(listener.heartbeatRan)
         assertNull(listener.heartbeatRunner)
         assertNull(listener.currentJob)
@@ -98,6 +90,9 @@ class TaskListenerTest {
         assertNull(listener.reporter)
     }
 
+    // ---------------------------------------------------------
+    // 2 — Heartbeat blokkerer ikke annen jobb
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når heartbeat kjører i bakgrunnen
@@ -110,62 +105,51 @@ class TaskListenerTest {
 
         val listener = object : TaskListener() {
 
-            var heartbeatStarted: Job? = null
             var heartbeatRan = false
 
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
-            override suspend fun onTask(task: Task): Event? {
+            override suspend fun onTask(task: Task): Event {
-
-                // Start heartbeat
                 withHeartbeatRunner(10.milliseconds) {
                     heartbeatRan = true
-                }.also { heartbeatStarted = it }
+                }
 
-                // Simuler annen coroutine-oppgave (VideoTaskListener/Converter)
                 launch {
                     delay(30)
                     otherWorkCompleted.complete(Unit)
                 }
 
-                // ⭐ Ikke fullfør onTask før testen sier det
                 allowFinish.await()
                 return object : Event() {}
             }
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
+        listener.accept(FakeTask(), reporter)
 
-        listener.accept(task, reporter)
-
-        // Vent på annen jobb
         otherWorkCompleted.await()
 
-        // ⭐ Nå er onTask fortsatt i live, cleanup har ikke skjedd
+        assertTrue(listener.heartbeatRan)
         assertNotNull(listener.currentJob)
         assertTrue(listener.currentJob!!.isActive)
 
-        // Heartbeat kjørte
-        assertNotNull(listener.heartbeatStarted)
-        assertTrue(listener.heartbeatRan)
-
-        // ⭐ Nå lar vi onTask fullføre
         allowFinish.complete(Unit)
+        listener.currentJob?.join()
 
-        // Vent på listener-jobben
-        listener.currentJob!!.join()
-
-        // Heartbeat ble kansellert
-        assertFalse(listener.heartbeatStarted!!.isActive)
-
-        // Cleanup
         assertNull(listener.heartbeatRunner)
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
     }
 
+    // ---------------------------------------------------------
+    // 3 — Heartbeat + CPU + IO arbeid
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når heartbeat kjører og flere parallelle jobber startes
@@ -179,73 +163,56 @@ class TaskListenerTest {
 
         val listener = object : TaskListener() {
 
-            var heartbeatStarted: Job? = null
             var heartbeatRan = false
 
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
             override suspend fun onTask(task: Task): Event? {
-
-                // Start heartbeat
                 withHeartbeatRunner(10.milliseconds) {
                     heartbeatRan = true
-                }.also { heartbeatStarted = it }
+                }
 
-                // Simuler Converter (CPU)
                 launch(Dispatchers.Default) {
-                    repeat(1000) { /* CPU work */ }
+                    repeat(1000) {}
                     converterDone.complete(Unit)
                 }
 
-                // Simuler VideoTaskListener (IO)
                 launch(Dispatchers.IO) {
                     delay(40)
                     videoDone.complete(Unit)
                 }
 
-                // ⭐ Vent til testen sier "nå kan du fullføre"
                 allowFinish.await()
                 return object : Event() {}
             }
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
+        listener.accept(FakeTask(), reporter)
 
-        listener.accept(task, reporter)
-
-        // Vent på begge "andre" oppgaver
         converterDone.await()
         videoDone.await()
 
-        // ⭐ Verifiser at begge faktisk ble fullført
-        assertTrue(converterDone.isCompleted)
-        assertTrue(videoDone.isCompleted)
-
-        // ⭐ Nå er onTask fortsatt i live, cleanup har ikke skjedd
-        assertNotNull(listener.currentJob)
-        assertTrue(listener.currentJob!!.isActive)
-
-        // Heartbeat kjørte
-        assertNotNull(listener.heartbeatStarted)
         assertTrue(listener.heartbeatRan)
+        assertNotNull(listener.currentJob)
 
-        // ⭐ Nå lar vi onTask fullføre
         allowFinish.complete(Unit)
+        listener.currentJob?.join()
 
-        // Vent på listener-jobben
-        listener.currentJob!!.join()
-
-        // Heartbeat ble kansellert
-        assertFalse(listener.heartbeatStarted!!.isActive)
-
-        // Cleanup
         assertNull(listener.heartbeatRunner)
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
     }
 
+    // ---------------------------------------------------------
+    // 4 — Arbeid fullføres, heartbeat kjører
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når onTask gjør ferdig arbeid
@@ -257,24 +224,25 @@ class TaskListenerTest {
 
         val listener = object : TaskListener() {
 
-            var heartbeatStarted: Job? = null
             var heartbeatRan = false
             var onTaskCalled = false
 
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
-            override suspend fun onTask(task: Task): Event? {
+            override suspend fun onTask(task: Task): Event {
                 onTaskCalled = true
 
                 withHeartbeatRunner(10.milliseconds) {
                     heartbeatRan = true
-                }.also { heartbeatStarted = it }
+                }
 
-                // Simuler arbeid
                 delay(20)
-
-                // ⭐ signaliser at arbeidet er ferdig
                 workCompleted.complete(Unit)
 
                 return object : Event() {}
@@ -282,34 +250,23 @@ class TaskListenerTest {
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
+        listener.accept(FakeTask(), reporter)
 
-        val accepted = listener.accept(task, reporter)
-        assertTrue(accepted)
-
-        // ⭐ Verifiser at arbeidet faktisk ble fullført
         workCompleted.await()
+        listener.currentJob?.join()
 
-        // Vent på jobben
-        listener.currentJob!!.join()
-
-        // onTask ble kalt
         assertTrue(listener.onTaskCalled)
-
-        // Heartbeat ble startet
-        assertNotNull(listener.heartbeatStarted)
         assertTrue(listener.heartbeatRan)
 
-        // Heartbeat ble kansellert
-        assertFalse(listener.heartbeatStarted!!.isActive)
-
-        // Cleanup
         assertNull(listener.heartbeatRunner)
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
         assertNull(listener.reporter)
     }
 
+    // ---------------------------------------------------------
+    // 5 — accept() returnerer false når busy
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når listener er opptatt med en task
@@ -321,36 +278,34 @@ class TaskListenerTest {
 
         val listener = object : TaskListener() {
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
             override suspend fun onTask(task: Task): Event? {
-                // Hold jobben i live
                 allowFinish.await()
                 return object : Event() {}
             }
         }
 
         val reporter = FakeReporter()
-        val task1 = FakeTask()
-        val task2 = FakeTask()
 
-        // Første task aksepteres
+        assertTrue(listener.accept(FakeTask(), reporter))
-        val accepted1 = listener.accept(task1, reporter)
+        assertFalse(listener.accept(FakeTask(), reporter))
-        assertTrue(accepted1)
 
-        // Listener er busy → andre task skal avvises
-        val accepted2 = listener.accept(task2, reporter)
-        assertFalse(accepted2)
-
-        // Fullfør første task
         allowFinish.complete(Unit)
-        listener.currentJob!!.join()
+        listener.currentJob?.join()
 
-        // Cleanup
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
     }
 
+    // ---------------------------------------------------------
+    // 6 — accept() returnerer false når unsupported
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når supports() returnerer false
@@ -360,21 +315,26 @@ class TaskListenerTest {
     fun acceptReturnsFalseWhenUnsupported() = runTest {
         val listener = object : TaskListener() {
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = false
             override suspend fun onTask(task: Task): Event? = error("Should not be called")
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
 
-        val accepted = listener.accept(task, reporter)
+        assertFalse(listener.accept(FakeTask(), reporter))
-
-        assertFalse(accepted)
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
         assertNull(listener.reporter)
     }
 
+    // ---------------------------------------------------------
+    // 7 — onError kalles når onTask kaster
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når onTask kaster en exception
@@ -386,6 +346,11 @@ class TaskListenerTest {
 
         val listener = object : TaskListener() {
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
             override suspend fun onTask(task: Task): Event? {
@@ -399,22 +364,19 @@ class TaskListenerTest {
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
+        listener.accept(FakeTask().newReferenceId(), reporter)
 
-        listener.accept(task, reporter)
-
-        // Vent på error-path
         errorLogged.await()
-
-        // ⭐ Vent på at cleanup i finally kjører
         listener.currentJob?.join()
 
-        // Cleanup verifisering
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
         assertNull(listener.heartbeatRunner)
     }
 
+    // ---------------------------------------------------------
+    // 8 — onCancelled kalles når jobben kanselleres
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når jobben kanselleres mens onTask kjører
@@ -427,11 +389,16 @@ class TaskListenerTest {
 
         val listener = object : TaskListener() {
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
             override suspend fun onTask(task: Task): Event? {
                 allowStart.complete(Unit)
-                delay(Long.MAX_VALUE) // hold jobben i live
+                delay(Long.MAX_VALUE)
                 return null
             }
 
@@ -442,28 +409,22 @@ class TaskListenerTest {
         }
 
         val reporter = FakeReporter()
-        val task = FakeTask()
+        listener.accept(FakeTask().newReferenceId(), reporter)
 
-        listener.accept(task, reporter)
-
-        // Vent til onTask har startet
         allowStart.await()
-
-        // Kanseller jobben
         listener.currentJob!!.cancel()
 
-        // Vent til onCancelled() ble kalt
         cancelledCalled.await()
-
-        // ⭐ Vent til cleanup i finally har kjørt
         listener.currentJob?.join()
 
-        // Cleanup verifisering
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
         assertNull(listener.heartbeatRunner)
     }
 
+    // ---------------------------------------------------------
+    // 9 — Sekvensiell kjøring uten state‑lekkasje
+    // ---------------------------------------------------------
     @Test
     @DisplayName("""
     Når listener prosesserer to tasks sekvensielt
@@ -471,7 +432,10 @@ class TaskListenerTest {
     Så skal ingen state lekke mellom tasks
     """)
     fun listenerHandlesSequentialTasksWithoutLeakingState() = runTest {
+        val started1 = CompletableDeferred<Unit>()
         val finish1 = CompletableDeferred<Unit>()
+
+        val started2 = CompletableDeferred<Unit>()
         val finish2 = CompletableDeferred<Unit>()
 
         val listener = object : TaskListener() {
@@ -479,41 +443,50 @@ class TaskListenerTest {
             var callCount = 0
 
             override fun getWorkerId() = "worker"
+
+            override fun createIncompleteStateTaskEvent(
+                task: Task, status: TaskStatus, exception: Exception?
+            ) = object : Event() {}
+
             override fun supports(task: Task) = true
 
-            override suspend fun onTask(task: Task): Event? {
+            override suspend fun onTask(task: Task): Event {
                 callCount++
-                if (callCount == 1) finish1.await()
+
-                if (callCount == 2) finish2.await()
+                if (callCount == 1) {
+                    started1.complete(Unit)
+                    finish1.await()
+                }
+
+                if (callCount == 2) {
+                    started2.complete(Unit)
+                    finish2.await()
+                }
+
                 return object : Event() {}
             }
         }
 
         val reporter = FakeReporter()
 
-        // Task 1
+        listener.accept(FakeTask(), reporter)
-        val task1 = FakeTask()
+        started1.await()
-        listener.accept(task1, reporter)
         finish1.complete(Unit)
-        listener.currentJob!!.join()
+        listener.currentJob?.join()
 
-        // Verifiser cleanup
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
         assertNull(listener.heartbeatRunner)
 
-        // Task 2
+        listener.accept(FakeTask(), reporter)
-        val task2 = FakeTask()
+        started2.await()
-        listener.accept(task2, reporter)
         finish2.complete(Unit)
-        listener.currentJob!!.join()
+        listener.currentJob?.join()
 
-        // Verifiser cleanup igjen
         assertNull(listener.currentJob)
         assertNull(listener.currentTask)
         assertNull(listener.heartbeatRunner)
 
-        // onTask ble kalt to ganger
         assertEquals(2, listener.callCount)
     }
 }

src/test/kotlin/no/iktdev/eventi/tasks/TaskPollerImplementationTest.kt

@@ -10,6 +10,7 @@ import no.iktdev.eventi.TestBase
 import no.iktdev.eventi.events.EventTypeRegistry
 import no.iktdev.eventi.models.Event
 import no.iktdev.eventi.models.Task
+import no.iktdev.eventi.models.store.TaskStatus
 import no.iktdev.eventi.stores.TaskStore
 import no.iktdev.eventi.testUtil.multiply
 import no.iktdev.eventi.testUtil.wipe
@@ -45,8 +46,8 @@ class TaskPollerImplementationTest : TestBase() {
             override fun markClaimed(taskId: UUID, workerId: String) {}
             override fun updateLastSeen(taskId: UUID) {}
             override fun markCompleted(taskId: UUID) {}
-            override fun markFailed(taskId: UUID) {}
+            override fun markFailed(referenceId: UUID,taskId: UUID) {}
-            override fun markCancelled(taskId: UUID) {}
+            override fun markCancelled(referenceId: UUID,taskId: UUID) {}
             override fun updateProgress(taskId: UUID, progress: Int) {}
             override fun log(taskId: UUID, message: String) {}
             override fun publishEvent(event: Event) {
@@ -72,6 +73,14 @@ class TaskPollerImplementationTest : TestBase() {
 
         fun getJob() = currentJob
         override fun getWorkerId() = this.javaClass.simpleName
+        override fun createIncompleteStateTaskEvent(
+            task: Task,
+            status: TaskStatus,
+            exception: Exception?
+        ): Event {
+            return object : Event() {}
+        }
+
         override fun supports(task: Task) = task is EchoTask
 
         override suspend fun onTask(task: Task): Event {
@@ -105,7 +114,7 @@ class TaskPollerImplementationTest : TestBase() {
         val listener = EchoListener()
         val poller = object : TaskPollerImplementation(taskStore, reporterFactory) {}
 
-        val task = EchoTask("Hello").newReferenceId().derivedOf(object : Event() {})
+        val task = EchoTask("Hello").newReferenceId().derivedOf(object : Event() {}.apply { newReferenceId() })
         taskStore.persist(task)
 
         poller.pollOnce()

src/test/kotlin/no/iktdev/eventi/testUtil/EventListenerRegistryUtil.kt

@@ -13,6 +13,7 @@ fun EventListenerRegistry.wipe() {
 
     // Tøm map’en
     val mutableList = field.get(EventListenerRegistry) as MutableList<*>
+    @Suppress("UNCHECKED_CAST")
     (mutableList as MutableList<Class<out EventListener>>).clear()
 
     // Verifiser at det er tomt

src/test/kotlin/no/iktdev/eventi/testUtil/EventTypeRegistryUtil.kt

@@ -13,6 +13,7 @@ fun EventTypeRegistry.wipe() {
 
     // Tøm map’en
     val typesMap = field.get(EventTypeRegistry) as MutableMap<*, *>
+    @Suppress("UNCHECKED_CAST")
     (typesMap as MutableMap<String, Class<out Event>>).clear()
 
     // Verifiser at det er tomt

src/test/kotlin/no/iktdev/eventi/testUtil/TaskListenerRegistryUtil.kt

@@ -1,7 +1,5 @@
 package no.iktdev.eventi.testUtil
 
-import no.iktdev.eventi.events.EventListener
-import no.iktdev.eventi.events.EventListenerRegistry
 import no.iktdev.eventi.tasks.TaskListener
 import no.iktdev.eventi.tasks.TaskListenerRegistry
 import org.assertj.core.api.Assertions.assertThat
@@ -15,6 +13,7 @@ fun TaskListenerRegistry.wipe() {
 
     // Tøm map’en
     val mutableList = field.get(TaskListenerRegistry) as MutableList<*>
+    @Suppress("UNCHECKED_CAST")
     (mutableList as MutableList<Class<out TaskListener>>).clear()
 
     // Verifiser at det er tomt

src/test/kotlin/no/iktdev/eventi/testUtil/TaskTypeRegistryUtil.kt

@@ -1,12 +1,11 @@
 package no.iktdev.eventi.testUtil
 
-import no.iktdev.eventi.events.EventTypeRegistry
-import no.iktdev.eventi.models.Event
 import no.iktdev.eventi.models.Task
 import no.iktdev.eventi.tasks.TaskTypeRegistry
 import org.junit.jupiter.api.Assertions.assertNull
 import java.lang.reflect.Field
 
+@Suppress("UNUSED_RECEIVER_PARAMETER")
 fun TaskTypeRegistry.wipe() {
     val field: Field = TaskTypeRegistry::class.java
         .superclass
@@ -15,6 +14,7 @@ fun TaskTypeRegistry.wipe() {
 
     // Tøm map’en
     val typesMap = field.get(TaskTypeRegistry) as MutableMap<*, *>
+    @Suppress("UNCHECKED_CAST")
     (typesMap as MutableMap<String, Class<out Task>>).clear()
 
     // Verifiser at det er tomt

src/test/kotlin/no/iktdev/eventi/testUtil/TestSequenceDispatchQueue.kt

@@ -0,0 +1,14 @@
+package no.iktdev.eventi.testUtil
+
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.SupervisorJob
+import no.iktdev.eventi.events.SequenceDispatchQueue
+
+class TestSequenceDispatchQueue(
+    maxConcurrency: Int,
+    dispatcher: CoroutineDispatcher
+) : SequenceDispatchQueue(
+    maxConcurrency,
+    CoroutineScope(dispatcher + SupervisorJob())
+)