From edffecbb3b52a66ca00f4714c099c9981ecd43ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tayfun=20Y=C4=B1lmaz?= Date: Sun, 13 Sep 2026 00:19:08 +0300 Subject: [PATCH 1/2] test(chain-busy): pin that a failed post-commit forward releases the chain reserve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends ChainBusyAcceptTests with the other half of the accept-time reserve: what happens when the forward it was taken for never succeeds. 'auto-leaf-to-waiting' exists in the leaf's definition but is not available in 'leaf-waiting'. The root has no such transition, so it accepts the request, marks the chain Busy down to the leaf and forwards — and the leaf rejects it with Transition:100020 (Validation). The post-commit failure policy treats that as the client's error and declines to fault, so neither settlement nor fault runs. Before the runtime fix nothing undid the reservation and every level stayed Busy permanently, with no incident and no retry path. The test pins three things: the observed status returns to Active so the client can act again; the ancestors stay Busy because each legitimately holds an open SubFlow correlation and must not be settled mid-subflow; and the chain still works afterwards, reaching root-done. Differential run (2026-09-12, local stack): with the runtime fix stashed and the hosts rebuilt the test fails at 31s with the chain still Busy; restored, the whole ChainBusy suite passes 15/15. Council 2026-09-08-parent-notification-mechanism, P0 (E31). Co-Authored-By: Claude Opus 5 --- TEST-SCENARIOS.md | 2 +- .../Tests/ChainBusy/ChainBusyAcceptTests.cs | 53 +++++++++++++++++++ .../Tests/ChainBusy/README.md | 32 ++++++++++- 3 files changed, 85 insertions(+), 2 deletions(-) diff --git a/TEST-SCENARIOS.md b/TEST-SCENARIOS.md index fc29f30..a147a7e 100644 --- a/TEST-SCENARIOS.md +++ b/TEST-SCENARIOS.md @@ -16,7 +16,7 @@ geçerliliğini yitiren senaryo **silinmez**, `deprecated` işaretlenip sebebi y | Senaryo | Test Edilen vNext Feature Seti | Neden Eklendi | Integration Test | Python Test | Durum | |---|---|---|---|---|---| | **error-boundary-lab** | Error boundary çözümlemesi: `CompiledBoundaryChain` Task→State→Global seviye baskınlığı · seviye içi sıra (`EffectivePriority` ASC → specificity DESC; default wildcard'ın 999'a düşmesi) · aksiyonlar abort/retry/rollback/notify/ignore/log · `TaskExecutionEngine` retry döngüsü (`1 + maxRetries`) ve tükenince `ResolveExcluding(Retry)` fallback'i · `BoundaryOutcomeHandler` → fault vs `RequestNextTransition` · `FinalizeTransitionStep`'in boundary transition'ı bitince incident'ı resolve etmesi · `InstanceIncidents` tablosu + denormalize `HasActiveIncident` · state function `incident` bloğu (link tabanlı, ResponseShapeVersion v9, ETag materyali) · `GET .../instances/{id}/incidents/active` (404 = açık arıza yok) · `GET .../instances/{id}/incidents` sayfalama · `metadata.incident` (state bloğuyla aynı şekil) · `POST .../retry` (400 `Instance:100027`, yeniden fault, veriyle kurtarma) · `queryRoles` kapısının incident geçmişine de uygulanması | vnext `feature/incident-table` (issue #865) incident'ları jsonb'den kendi tablosuna taşıdı ve iki yeni client yüzeyi ekledi; ayrıca error boundary bu repoda hiç senaryo olarak yoktu ve `POST .../retry` hiçbir testte çağrılmıyordu (2026-09-06) | `Tests/ErrorBoundaryLab` (6 sınıf, 31 test) | — (bilinçli: davranış/çözümleme senaryosu, eşzamanlılık iddiası yok) | ✅ **Aktif — 31/31 yeşil** (lokal runtime `feature/incident-table`, art arda iki koşu, ~1 dk 46 sn; ilk koşu 2026-09-06, üç davranışsal kusurun düzeltilmesinden sonra 2026-09-07'de yeniden) | -| **chain-busy** | Accept-time subflow chain reserve · Busy-as-mutex · `$self` shared transition vs `updateData` lifecycle sınırı · start `initial → initial` semantiği · cancel propagasyonu (in-process ↕ distributed) · scheduled transition re-arm | `updateData`-only self-target profil sınırını pinlemek — `target: $self` "hook'ları atla" demek değil (2026-08-17) | `Tests/ChainBusy` (5 sınıf) | `api-tests/chain-busy/chain-busy-behaviour-test.py`, `chain-busy-accept-test.py` | ✅ Aktif | +| **chain-busy** | Accept-time subflow chain reserve **ve başarısız forward'da geri alınması (E31)** · Busy-as-mutex · `$self` shared transition vs `updateData` lifecycle sınırı · start `initial → initial` semantiği · cancel propagasyonu (in-process ↕ distributed) · scheduled transition re-arm | `updateData`-only self-target profil sınırını pinlemek — `target: $self` "hook'ları atla" demek değil (2026-08-17). **E31 (2026-09-12):** post-commit forward istemci hatasıyla düşünce rezervasyon salınmıyordu; üretimde 30 günde 51 mahsur instance (council `2026-09-08-parent-notification-mechanism`, P0) | `Tests/ChainBusy` (5 sınıf) | `api-tests/chain-busy/chain-busy-behaviour-test.py`, `chain-busy-accept-test.py` | ✅ Aktif | | **script-race-lab** | Script engine: paylaşılan `AssemblyLoadContext`'te çift-derleme yarışı · `scripts.helpers` · subflow output mapping · parent kalıcı fault riski | `Script_XXXX already loaded` / `FileLoadException` yarışının fixture'ı; fix'siz runtime'da kaybedenler parent'ı kalıcı fault'lar (2026-08-18) | `Tests/ScriptRaceLab` | `api-tests/script-race-lab/race-load.py` (yük), `publish.py` | ✅ Aktif | | **data-integrity-lab** | InstanceData v2: anında persist · lock altında kimlik · sıralı/paralel task yazımları · DataHash dedup (task + updateData) · versiyon satırı bütünlüğü | `feature/busy-as-mutex-locking` + InstanceData v2 geliştirmesini uçtan uca ölçmek (2026-08-13) | `Tests/DataIntegrityLab` | `api-tests/data-integrity-lab/integrity-lab-test.py` | ⚠️ Kısmen kırmızı — `run-parallel` konteynerli ortamda settle olmuyor (120s'te doğrulandı, hang) | | **subflow-orchestration** | 3 seviyeli subflow (parent → child → grandchild) · `updateData`'nın auto transition'ı tetiklemesi · aktif subflow'lu parent'ta `updateData` data-only kısa devresi · parent `$self` shared transition · eşzamanlı updateData tutarlılığı | Platformun subflow referans akışı; `feature/busy-as-mutex-locking` F1/F1a/F8 fix'lerinin doğrulaması (2026-04-28) | `Tests/SubflowOrchestration` | `api-tests/subflow-orchestration/updatedata-concurrency-test.py` | ✅ Aktif | diff --git a/tests/Core.IntegrationTests/Tests/ChainBusy/ChainBusyAcceptTests.cs b/tests/Core.IntegrationTests/Tests/ChainBusy/ChainBusyAcceptTests.cs index 2649ce7..aacc3b9 100644 --- a/tests/Core.IntegrationTests/Tests/ChainBusy/ChainBusyAcceptTests.cs +++ b/tests/Core.IntegrationTests/Tests/ChainBusy/ChainBusyAcceptTests.cs @@ -35,6 +35,59 @@ public async Task AsyncAccept_MarksTheChainBusyDownToTheLeaf_BeforeAnsweringTheC Assert.Equal("B", after.Status); } + [Fact] + public async Task PostCommitForwardFailure_ReleasesTheChainReserve_SoNoLevelStaysBusy() + { + // E31 (council 2026-09-08, P0). The other half of the reserve: what happens when the + // forward it was taken for never succeeds. + // + // 'auto-leaf-to-waiting' exists in the leaf's definition but is not available in + // 'leaf-waiting'. The root has no such transition, so it accepts the request, marks the + // chain Busy down to the leaf and forwards — and the leaf rejects it with a Validation + // error. The post-commit failure policy classifies that as the client's error, so it + // declines to fault; that exit runs neither settlement nor fault, and before the fix + // nothing undid the reservation. Every level stayed Busy permanently: Busy has no + // recovery API (retry requires Faulted) and no incident is raised, so the instance was + // unreachable without direct database intervention. Measured in production: 51 stranded + // instances in 30 days, across all five live runtime versions. + var chain = await BuildChainAsync("reserve-release"); + + var (status, _) = await RunTransitionAsyncModeAsync( + RootWorkflow, chain.RootId, "auto-leaf-to-waiting"); + Assert.True(status is HttpStatusCode.Accepted or HttpStatusCode.OK, + $"the accept was rejected with {status} — the reserve was never taken, so this test " + + "would pass without exercising the compensation"); + + // The client polls the root and only ever sees the leaf. That is the observation the + // compensation has to restore; before the fix it read "B" forever. + await WaitUntilAsync( + async () => (await GetObservedStateAsync(RootWorkflow, chain.RootId)).Status == "A", + $"the chain stayed Busy after the forward failed — the accept-time reserve was not " + + $"released. root={await DescribeAsync(RootWorkflow, chain.RootId)}", + TimeSpan.FromSeconds(30)); + + var observed = await GetObservedStateAsync(RootWorkflow, chain.RootId); + Assert.Equal(LeafRestingState, observed.State); + + // The leaf owns its own visible status, so the release flips it back. + var leaf = await GetInstanceStateAsync(LeafWorkflow, chain.LeafId); + Assert.Equal("A", leaf.Status); + Assert.Equal(LeafRestingState, leaf.State); + + // The ancestors must NOT be released: each holds an open SubFlow correlation and is + // legitimately Busy for its child's whole lifetime. Releasing them would settle a parent + // that is still mid-subflow — the compensation undoes only what the reserve flipped. + Assert.Equal("B", (await GetInstanceStateAsync(RootWorkflow, chain.RootId)).Status); + Assert.Equal("B", (await GetInstanceStateAsync(MiddleWorkflow, chain.MiddleId)).Status); + + // Not merely un-Busied: the chain still works. This is what "stranded" cost in + // production — the flow could never be driven to completion again. + await RunTransitionAsyncModeAsync(RootWorkflow, chain.RootId, "finish-leaf"); + await WaitUntilTerminalAsync(RootWorkflow, chain.RootId, TimeSpan.FromSeconds(90)); + + Assert.Equal("root-done", (await GetInstanceStateAsync(RootWorkflow, chain.RootId)).State); + } + [Fact] public async Task AsyncAccept_RelaysTheTransitionAllTheWayToTheLeaf() { diff --git a/tests/Core.IntegrationTests/Tests/ChainBusy/README.md b/tests/Core.IntegrationTests/Tests/ChainBusy/README.md index be186e6..fb7923b 100644 --- a/tests/Core.IntegrationTests/Tests/ChainBusy/README.md +++ b/tests/Core.IntegrationTests/Tests/ChainBusy/README.md @@ -28,11 +28,41 @@ erişimi gerekmez. | Sınıf | Doğruladığı | | --- | --- | | `ChainBusyStartTests` | Start, başlangıç state'inin onEntry'sini çalıştırır (üst seviye + subflow); zincir kurulduğunda A/B `Busy`, C `Active`. | -| `ChainBusyAcceptTests` | Async accept, 202 dönmeden **önce** zinciri leaf'e kadar `Busy`'e çeker; relay gerçekten leaf'e ulaşır. | +| `ChainBusyAcceptTests` | Async accept, 202 dönmeden **önce** zinciri leaf'e kadar `Busy`'e çeker; relay gerçekten leaf'e ulaşır; **forward başarısız olduğunda rezervasyon geri alınır** (E31, aşağı bak). | | `ChainBusySharedTransitionTests` | `$self` shared transition kendi işini yapar **ve** state yaşam döngüsünü koşar (onEntry/onExit girer, zamanlayıcıyı yeniden kurar) — `target: $self` "instance'ı oynatma" der, "hook'ları atla" demez; parent'ın kendi shared'ı parent'ta karşılanır (forward edilmez); yalnız leaf'te tanımlı olan aşağı forward edilir. | | `ChainBusyUpdateDataTests` | `updateData` onEntry/onExit çalıştırmaz, zamanlayıcıyı yeniden kurmaz, state'i değiştirmez. Yaşam döngüsü atlamasını alan **tek** transition; `ChainBusySharedTransitionTests` ile birlikte sınırı pinler — birini diğeri olmadan değiştirmek sınırı sessizce siler. | | `ChainBusyCancelTests` | Leaf'ten cancel → yukarı tamamlanma + korelasyon kapanır. Root'tan cancel → aşağı kaskad. | +## E31 — forward başarısız olduğunda zincir rezervasyonu + +`PostCommitForwardFailure_ReleasesTheChainReserve_SoNoLevelStaysBusy` accept-time rezervasyonun +diğer yarısını pinler: uğruna alındığı forward hiç başarılı olmazsa ne olacağı. + +`auto-leaf-to-waiting` leaf'in tanımında vardır ama `leaf-waiting` state'inde **kullanılabilir +değildir**. Root'ta böyle bir transition olmadığı için root isteği kabul eder, zinciri leaf'e kadar +`Busy` damgalar ve aşağı iletir; leaf de onu `Transition:100020` (Validation) ile reddeder. +Post-commit hata politikası Validation'ı **istemci hatası** sayıp fault'lamayı reddeder — ve o çıkış +ne settlement ne fault koşturur. Düzeltmeden önce rezervasyonu geri alan hiçbir şey yoktu. + +Sonuç kalıcı bir mahsur kalmaydı: `Busy`'nin kurtarma API'si yok (retry `Faulted` ister), incident +de açılmıyor (`hasActiveIncident: false`), dolayısıyla instance veritabanına elle müdahale +edilmeden erişilemez hale geliyordu. Üretimde ölçülen: **30 günde 51 mahsur instance**, beş canlı +runtime sürümünün hepsinde (agent council `2026-09-08-parent-notification-mechanism`, P0). + +Testin pinlediği üç şey: + +1. **Client yeniden ilerleyebilir** — root'tan gözlenen durum `A`'ya döner (düzeltmeden önce + süresiz `B` okunuyordu; ölçüldü: 90 saniye sonra hâlâ `B`). +2. **Atalar salınmaz** — root ve middle `B` kalır. Her biri açık SubFlow korelasyonu tutar ve + çocuğunun ömrü boyunca meşru olarak `Busy`'dir; telafi yalnızca rezervasyonun çevirdiğini geri + alır, hâlâ subflow ortasında olan bir parent'ı settle etmez. +3. **Akış gerçekten kullanılabilir** — ardından `finish-leaf` çalışır ve zincir `root-done`'a ulaşır. + Mahsur kalmanın üretimdeki bedeli buydu: akış bir daha asla tamamlanamıyordu. + +Ayırt edici koşu (2026-09-12, lokal stack): düzeltme `git stash` ile çıkarılıp host'lar yeniden +derlendiğinde test düşer (31 sn, zincir `B`'de kalır); geri konduğunda ChainBusy takımının tamamı +(15 test) geçer. + ## Çalıştırma ### Konteynerli ortam (varsayılan) From ce377a1a1c0489fdd37ec4eac040b446454e03a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tayfun=20Y=C4=B1lmaz?= Date: Sun, 13 Sep 2026 09:09:35 +0300 Subject: [PATCH 2/2] test(event-driven-lab): cover the events endpoint over a real broker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The POST .../instances/events endpoint had no integration test of any kind — recorded as a named gap by agent council 2026-09-11-route-trace-coverage. A test that posted to the endpoint directly would prove the mapping script and nothing about the wiring that actually delivers production traffic: the topic, the subscription route, the CloudEvent unwrap and the triggerType 3 gate. So both messages go through the broker, and the test never calls the runtime's API to drive the flow. The instance is created by an event and advanced by an event, correlated by business key alone — no instance id is ever published. The flow follows the documented configuration rather than an invented one: a workflow-level event mapping for action=start, a triggerType 3 transition with its own mapping for action=transition, and two Dapr Subscriptions whose routes carry action and transitionKey. The approve subscription uses sync=true so the delivery blocks until the pipeline reaches a rest point. Three things are pinned, including the one a state-only assertion would miss: the transition must run with the MAPPED body, not merely move the state. And an approval naming no active instance must be acked rather than retried — the endpoint answers the broker's protocol, so "nothing matched" must not look like a failure or the broker redelivers a message this runtime can never satisfy. The README records the trap that cost time here: Dapr reads declarative subscriptions only at startup, so a subscription added while the sidecar is running sends messages nowhere, with no error and no log. Co-Authored-By: Claude Opus 5 --- TEST-SCENARIOS.md | 1 + .../event-lab-script-task.json | 16 ++ .../build-event-driven-lab.py | 104 +++++++++++ .../event-driven-lab/event-driven-lab.json | 113 ++++++++++++ .../src/ApproveEventMapping.csx | 28 +++ .../src/RecordApprovalMapping.csx | 25 +++ .../src/StartEventMapping.csx | 36 ++++ ...subscription-event-driven-lab-approve.yaml | 16 ++ .../subscription-event-driven-lab-start.yaml | 15 ++ .../EventDrivenLab/EventDrivenLabTests.cs | 167 ++++++++++++++++++ .../Tests/EventDrivenLab/README.md | 59 +++++++ 11 files changed, 580 insertions(+) create mode 100644 core/Tasks/event-driven-lab/event-lab-script-task.json create mode 100644 core/Workflows/event-driven-lab/build-event-driven-lab.py create mode 100644 core/Workflows/event-driven-lab/event-driven-lab.json create mode 100644 core/Workflows/event-driven-lab/src/ApproveEventMapping.csx create mode 100644 core/Workflows/event-driven-lab/src/RecordApprovalMapping.csx create mode 100644 core/Workflows/event-driven-lab/src/StartEventMapping.csx create mode 100644 tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-approve.yaml create mode 100644 tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-start.yaml create mode 100644 tests/Core.IntegrationTests/Tests/EventDrivenLab/EventDrivenLabTests.cs create mode 100644 tests/Core.IntegrationTests/Tests/EventDrivenLab/README.md diff --git a/TEST-SCENARIOS.md b/TEST-SCENARIOS.md index a147a7e..17ec586 100644 --- a/TEST-SCENARIOS.md +++ b/TEST-SCENARIOS.md @@ -16,6 +16,7 @@ geçerliliğini yitiren senaryo **silinmez**, `deprecated` işaretlenip sebebi y | Senaryo | Test Edilen vNext Feature Seti | Neden Eklendi | Integration Test | Python Test | Durum | |---|---|---|---|---|---| | **error-boundary-lab** | Error boundary çözümlemesi: `CompiledBoundaryChain` Task→State→Global seviye baskınlığı · seviye içi sıra (`EffectivePriority` ASC → specificity DESC; default wildcard'ın 999'a düşmesi) · aksiyonlar abort/retry/rollback/notify/ignore/log · `TaskExecutionEngine` retry döngüsü (`1 + maxRetries`) ve tükenince `ResolveExcluding(Retry)` fallback'i · `BoundaryOutcomeHandler` → fault vs `RequestNextTransition` · `FinalizeTransitionStep`'in boundary transition'ı bitince incident'ı resolve etmesi · `InstanceIncidents` tablosu + denormalize `HasActiveIncident` · state function `incident` bloğu (link tabanlı, ResponseShapeVersion v9, ETag materyali) · `GET .../instances/{id}/incidents/active` (404 = açık arıza yok) · `GET .../instances/{id}/incidents` sayfalama · `metadata.incident` (state bloğuyla aynı şekil) · `POST .../retry` (400 `Instance:100027`, yeniden fault, veriyle kurtarma) · `queryRoles` kapısının incident geçmişine de uygulanması | vnext `feature/incident-table` (issue #865) incident'ları jsonb'den kendi tablosuna taşıdı ve iki yeni client yüzeyi ekledi; ayrıca error boundary bu repoda hiç senaryo olarak yoktu ve `POST .../retry` hiçbir testte çağrılmıyordu (2026-09-06) | `Tests/ErrorBoundaryLab` (6 sınıf, 31 test) | — (bilinçli: davranış/çözümleme senaryosu, eşzamanlılık iddiası yok) | ✅ **Aktif — 31/31 yeşil** (lokal runtime `feature/incident-table`, art arda iki koşu, ~1 dk 46 sn; ilk koşu 2026-09-06, üç davranışsal kusurun düzeltilmesinden sonra 2026-09-07'de yeniden) | +| **event-driven-lab** | Olay tetiklemeli start + `triggerType: 3` transition · Dapr Subscription route'u · CloudEvent açma · `IEventMapping` ile iş anahtarı korelasyonu · eşleşmeyen olayın ack'lenmesi | `POST .../instances/events` ucunun **hiç testi yoktu** (council `2026-09-11-route-trace-coverage`, adı konmuş boşluk); ayrıca `Event.Intake` span'i bu ucu ölçüyor ve testsiz route'a span gönderilmez (2026-09-13) | `Tests/EventDrivenLab` (1 sınıf) | — | ✅ Aktif | | **chain-busy** | Accept-time subflow chain reserve **ve başarısız forward'da geri alınması (E31)** · Busy-as-mutex · `$self` shared transition vs `updateData` lifecycle sınırı · start `initial → initial` semantiği · cancel propagasyonu (in-process ↕ distributed) · scheduled transition re-arm | `updateData`-only self-target profil sınırını pinlemek — `target: $self` "hook'ları atla" demek değil (2026-08-17). **E31 (2026-09-12):** post-commit forward istemci hatasıyla düşünce rezervasyon salınmıyordu; üretimde 30 günde 51 mahsur instance (council `2026-09-08-parent-notification-mechanism`, P0) | `Tests/ChainBusy` (5 sınıf) | `api-tests/chain-busy/chain-busy-behaviour-test.py`, `chain-busy-accept-test.py` | ✅ Aktif | | **script-race-lab** | Script engine: paylaşılan `AssemblyLoadContext`'te çift-derleme yarışı · `scripts.helpers` · subflow output mapping · parent kalıcı fault riski | `Script_XXXX already loaded` / `FileLoadException` yarışının fixture'ı; fix'siz runtime'da kaybedenler parent'ı kalıcı fault'lar (2026-08-18) | `Tests/ScriptRaceLab` | `api-tests/script-race-lab/race-load.py` (yük), `publish.py` | ✅ Aktif | | **data-integrity-lab** | InstanceData v2: anında persist · lock altında kimlik · sıralı/paralel task yazımları · DataHash dedup (task + updateData) · versiyon satırı bütünlüğü | `feature/busy-as-mutex-locking` + InstanceData v2 geliştirmesini uçtan uca ölçmek (2026-08-13) | `Tests/DataIntegrityLab` | `api-tests/data-integrity-lab/integrity-lab-test.py` | ⚠️ Kısmen kırmızı — `run-parallel` konteynerli ortamda settle olmuyor (120s'te doğrulandı, hang) | diff --git a/core/Tasks/event-driven-lab/event-lab-script-task.json b/core/Tasks/event-driven-lab/event-lab-script-task.json new file mode 100644 index 0000000..cf24c76 --- /dev/null +++ b/core/Tasks/event-driven-lab/event-lab-script-task.json @@ -0,0 +1,16 @@ +{ + "key": "event-lab-script-task", + "version": "1.0.0", + "domain": "core", + "flow": "sys-tasks", + "flowVersion": "1.0.0", + "tags": [ + "integration-test", + "event-driven-lab", + "script-task" + ], + "attributes": { + "type": "7", + "config": {} + } +} diff --git a/core/Workflows/event-driven-lab/build-event-driven-lab.py b/core/Workflows/event-driven-lab/build-event-driven-lab.py new file mode 100644 index 0000000..0cad2c0 --- /dev/null +++ b/core/Workflows/event-driven-lab/build-event-driven-lab.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Regenerates event-driven-lab.json from the .csx sources in ./src. + +Every mapping is embedded as base64 in `code` next to its `location`; edit the .csx files and +re-run this script — never hand-edit the base64 blobs. + + python3 core/Workflows/event-driven-lab/build-event-driven-lab.py +""" + +import base64 +import json +from pathlib import Path + +ROOT = Path(__file__).resolve().parent + + +def code(name): + raw = (ROOT / "src" / name).read_bytes() + return {"location": f"./src/{name}", "code": base64.b64encode(raw).decode()} + + +def label(text): + return [{"language": "en-US", "label": text}] + + +SCRIPT_TASK = { + "key": "event-lab-script-task", + "domain": "core", + "version": "1.0.0", + "flow": "sys-tasks", +} + +workflow = { + "key": "event-driven-lab", + "flow": "sys-flows", + "flowVersion": "1.0.0", + "domain": "core", + # Publishing is version-immutable: the same version with different content is refused with + # Instance:100002. Every content change here needs a patch bump. + "version": "1.0.1", + "tags": ["integration-test", "event-driven-lab", "event", "pubsub", "tracing"], + "attributes": { + "type": "F", + "labels": label("Event Driven Lab"), + "functions": [], + "extensions": [], + # Workflow-level event mapping = the action=start subscription's entry point. + "event": {"mapping": code("StartEventMapping.csx")}, + "startTransition": { + "key": "start-event-driven-lab", + "target": "awaiting-approval", + "triggerType": 0, + "versionStrategy": "Major", + "labels": label("Start Event Driven Lab"), + }, + "states": [ + { + "key": "awaiting-approval", + "stateType": 1, + "subType": 0, + "versionStrategy": "Major", + "labels": label("Awaiting Approval"), + "view": None, + "subFlow": None, + "onEntries": [], + "onExits": [], + "transitions": [ + { + "key": "approve-by-event", + "target": "approved", + # 3 = Event. Only an event delivery may drive this transition. + "triggerType": 3, + "versionStrategy": "Minor", + "labels": label("Approve By Event"), + "event": {"mapping": code("ApproveEventMapping.csx")}, + "onExecutionTasks": [ + { + "order": 1, + "task": SCRIPT_TASK, + "mapping": code("RecordApprovalMapping.csx"), + } + ], + } + ], + }, + { + "key": "approved", + "stateType": 3, + "subType": 1, + "versionStrategy": "Major", + "labels": label("Approved"), + "view": None, + "subFlow": None, + "onEntries": [], + "onExits": [], + "transitions": [], + }, + ], + }, +} + +out = ROOT / "event-driven-lab.json" +out.write_text(json.dumps(workflow, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") +print(f"wrote {out}") diff --git a/core/Workflows/event-driven-lab/event-driven-lab.json b/core/Workflows/event-driven-lab/event-driven-lab.json new file mode 100644 index 0000000..c82a607 --- /dev/null +++ b/core/Workflows/event-driven-lab/event-driven-lab.json @@ -0,0 +1,113 @@ +{ + "key": "event-driven-lab", + "flow": "sys-flows", + "flowVersion": "1.0.0", + "domain": "core", + "version": "1.0.1", + "tags": [ + "integration-test", + "event-driven-lab", + "event", + "pubsub", + "tracing" + ], + "attributes": { + "type": "F", + "labels": [ + { + "language": "en-US", + "label": "Event Driven Lab" + } + ], + "functions": [], + "extensions": [], + "event": { + "mapping": { + "location": "./src/StartEventMapping.csx", + "code": "dXNpbmcgU3lzdGVtLlRocmVhZGluZy5UYXNrczsKdXNpbmcgQkJULldvcmtmbG93LlNjcmlwdGluZzsKCi8vLyA8c3VtbWFyeT4KLy8vIFdvcmtmbG93LWxldmVsIGV2ZW50IG1hcHBpbmc6IHR1cm5zIGEgcmF3IGJyb2tlciBtZXNzYWdlIGludG8gYSBuZXcgaW5zdGFuY2UuCi8vLyA8cGFyYT4KLy8vIFJlYWNoZWQgdGhyb3VnaCA8Yz5QT1NUIC4uLi9pbnN0YW5jZXMvZXZlbnRzP2FjdGlvbj1zdGFydDwvYz4sIHdoaWNoIGEgRGFwciBTdWJzY3JpcHRpb24gcm91dGVzCi8vLyB0byBmcm9tIHRoZSB0b3BpYy4gVGhlIHJ1bnRpbWUgaGFzIGFscmVhZHkgdW53cmFwcGVkIHRoZSBDbG91ZEV2ZW50IGVudmVsb3BlLCBzbwovLy8gPGM+Y29udGV4dC5FdmVudFBheWxvYWQ8L2M+IGlzIHRoZSBtZXNzYWdlIHRoZSBwdWJsaXNoZXIgYWN0dWFsbHkgc2VudC4KLy8vIDwvcGFyYT4KLy8vIDxwYXJhPgovLy8gPGM+SW5zdGFuY2VLZXk8L2M+IGlzIHRoZSBjb3JyZWxhdGlvbiBjb250cmFjdCBmb3IgZXZlcnl0aGluZyB0aGF0IGZvbGxvd3M6IHRoZSBhcHByb3ZlIGV2ZW50Ci8vLyBsYXRlciBmaW5kcyB0aGlzIGluc3RhbmNlIGJ5IGV4YWN0bHkgdGhpcyB2YWx1ZSwgc28gaXQgbXVzdCBjb21lIGZyb20gdGhlIHBheWxvYWQgYW5kIG5vdCBiZQovLy8gZ2VuZXJhdGVkIGhlcmUuCi8vLyA8L3BhcmE+Ci8vLyA8L3N1bW1hcnk+CnB1YmxpYyBjbGFzcyBTdGFydEV2ZW50TWFwcGluZyA6IElFdmVudE1hcHBpbmcKewogICAgcHVibGljIGFzeW5jIFRhc2s8RXZlbnRNYXBwaW5nUmVzdWx0PiBIYW5kbGVyKFNjcmlwdENvbnRleHQgY29udGV4dCkKICAgIHsKICAgICAgICB2YXIgcGF5bG9hZCA9IGNvbnRleHQuRXZlbnRQYXlsb2FkOwoKICAgICAgICByZXR1cm4gYXdhaXQgVGFzay5Gcm9tUmVzdWx0KG5ldyBFdmVudE1hcHBpbmdSZXN1bHQKICAgICAgICB7CiAgICAgICAgICAgIEluc3RhbmNlS2V5ID0gKHN0cmluZylwYXlsb2FkLm9yZGVySWQsCiAgICAgICAgICAgIEJvZHkgPSBuZXcKICAgICAgICAgICAgewogICAgICAgICAgICAgICAgb3JkZXJJZCA9IChzdHJpbmcpcGF5bG9hZC5vcmRlcklkLAogICAgICAgICAgICAgICAgYW1vdW50ID0gcGF5bG9hZC5hbW91bnQsCiAgICAgICAgICAgICAgICAvLyBQcm92ZXMgdGhlIGluc3RhbmNlIHdhcyBjcmVhdGVkIGJ5IHRoZSBldmVudCBwYXRoIHJhdGhlciB0aGFuIGJ5IGFuIEhUVFAgc3RhcnQuCiAgICAgICAgICAgICAgICBzdGFydGVkQnkgPSAiZXZlbnQiLAogICAgICAgICAgICAgICAgYXBwcm92YWxzID0gMAogICAgICAgICAgICB9CiAgICAgICAgfSk7CiAgICB9Cn0K" + } + }, + "startTransition": { + "key": "start-event-driven-lab", + "target": "awaiting-approval", + "triggerType": 0, + "versionStrategy": "Major", + "labels": [ + { + "language": "en-US", + "label": "Start Event Driven Lab" + } + ] + }, + "states": [ + { + "key": "awaiting-approval", + "stateType": 1, + "subType": 0, + "versionStrategy": "Major", + "labels": [ + { + "language": "en-US", + "label": "Awaiting Approval" + } + ], + "view": null, + "subFlow": null, + "onEntries": [], + "onExits": [], + "transitions": [ + { + "key": "approve-by-event", + "target": "approved", + "triggerType": 3, + "versionStrategy": "Minor", + "labels": [ + { + "language": "en-US", + "label": "Approve By Event" + } + ], + "event": { + "mapping": { + "location": "./src/ApproveEventMapping.csx", + "code": "dXNpbmcgU3lzdGVtLlRocmVhZGluZy5UYXNrczsKdXNpbmcgQkJULldvcmtmbG93LlNjcmlwdGluZzsKCi8vLyA8c3VtbWFyeT4KLy8vIFRyYW5zaXRpb24tbGV2ZWwgZXZlbnQgbWFwcGluZyBmb3IgPGM+YXBwcm92ZS1ieS1ldmVudDwvYz4gKDxjPnRyaWdnZXJUeXBlOiAzPC9jPikuCi8vLyA8cGFyYT4KLy8vIENvcnJlbGF0ZXMgYnkgYnVzaW5lc3Mga2V5OiB0aGUgcnVudGltZSBsb29rcyB1cCB0aGUgQUNUSVZFIGluc3RhbmNlIHdob3NlIGtleSBlcXVhbHMKLy8vIDxjPkluc3RhbmNlS2V5PC9jPi4gTm8gbWF0Y2ggaXMgYSBub3JtYWwgYW5zd2VyIOKAlCB0aGUgZW5kcG9pbnQgYWNrcyBzbyB0aGUgYnJva2VyIGRvZXMgbm90Ci8vLyByZWRlbGl2ZXIgZm9yZXZlciDigJQgd2hpY2ggaXMgd2h5IHRoZSB0ZXN0IGFzc2VydHMgb24gdGhlIGluc3RhbmNlIHJhdGhlciB0aGFuIG9uIHRoZSByZXNwb25zZS4KLy8vIDwvcGFyYT4KLy8vIDwvc3VtbWFyeT4KcHVibGljIGNsYXNzIEFwcHJvdmVFdmVudE1hcHBpbmcgOiBJRXZlbnRNYXBwaW5nCnsKICAgIHB1YmxpYyBhc3luYyBUYXNrPEV2ZW50TWFwcGluZ1Jlc3VsdD4gSGFuZGxlcihTY3JpcHRDb250ZXh0IGNvbnRleHQpCiAgICB7CiAgICAgICAgdmFyIHBheWxvYWQgPSBjb250ZXh0LkV2ZW50UGF5bG9hZDsKCiAgICAgICAgcmV0dXJuIGF3YWl0IFRhc2suRnJvbVJlc3VsdChuZXcgRXZlbnRNYXBwaW5nUmVzdWx0CiAgICAgICAgewogICAgICAgICAgICBJbnN0YW5jZUtleSA9IChzdHJpbmcpcGF5bG9hZC5vcmRlcklkLAogICAgICAgICAgICBCb2R5ID0gbmV3CiAgICAgICAgICAgIHsKICAgICAgICAgICAgICAgIGRlY2lzaW9uID0gKHN0cmluZylwYXlsb2FkLmRlY2lzaW9uLAogICAgICAgICAgICAgICAgYXBwcm92ZWRCeSA9IChzdHJpbmcpcGF5bG9hZC5hcHByb3ZlZEJ5CiAgICAgICAgICAgIH0KICAgICAgICB9KTsKICAgIH0KfQo=" + } + }, + "onExecutionTasks": [ + { + "order": 1, + "task": { + "key": "event-lab-script-task", + "domain": "core", + "version": "1.0.0", + "flow": "sys-tasks" + }, + "mapping": { + "location": "./src/RecordApprovalMapping.csx", + "code": "dXNpbmcgU3lzdGVtLkNvbGxlY3Rpb25zLkdlbmVyaWM7CnVzaW5nIFN5c3RlbS5EeW5hbWljOwp1c2luZyBTeXN0ZW0uVGhyZWFkaW5nLlRhc2tzOwp1c2luZyBCQlQuV29ya2Zsb3cuRGVmaW5pdGlvbnM7CnVzaW5nIEJCVC5Xb3JrZmxvdy5TY3JpcHRpbmc7CgovLy8gPHN1bW1hcnk+Ci8vLyBXcml0ZXMgd2hhdCB0aGUgZXZlbnQgY2FycmllZCBpbnRvIGluc3RhbmNlIGRhdGEsIHNvIHRoZSB0ZXN0IGNhbiBhc3NlcnQgdGhhdCB0aGUgdHJhbnNpdGlvbiByYW4KLy8vIHdpdGggdGhlIG1hcHBlZCBib2R5IHJhdGhlciB0aGFuIG1lcmVseSB0aGF0IHRoZSBzdGF0ZSBtb3ZlZC4KLy8vIDwvc3VtbWFyeT4KcHVibGljIGNsYXNzIFJlY29yZEFwcHJvdmFsTWFwcGluZyA6IElNYXBwaW5nCnsKICAgIHB1YmxpYyBhc3luYyBUYXNrPFNjcmlwdFJlc3BvbnNlPiBJbnB1dEhhbmRsZXIoV29ya2Zsb3dUYXNrIHRhc2ssIFNjcmlwdENvbnRleHQgY29udGV4dCkKICAgIHsKICAgICAgICBkeW5hbWljIGJvZHkgPSBuZXcgRXhwYW5kb09iamVjdCgpOwogICAgICAgIGJvZHkuYXBwcm92YWxzID0gMTsKICAgICAgICBib2R5LmRlY2lzaW9uID0gY29udGV4dC5Cb2R5Py5kZWNpc2lvbjsKICAgICAgICBib2R5LmFwcHJvdmVkQnkgPSBjb250ZXh0LkJvZHk/LmFwcHJvdmVkQnk7CgogICAgICAgIHJldHVybiBhd2FpdCBUYXNrLkZyb21SZXN1bHQobmV3IFNjcmlwdFJlc3BvbnNlIHsgRGF0YSA9IGJvZHkgfSk7CiAgICB9CgogICAgcHVibGljIGFzeW5jIFRhc2s8U2NyaXB0UmVzcG9uc2U+IE91dHB1dEhhbmRsZXIoU2NyaXB0Q29udGV4dCBjb250ZXh0KQogICAgICAgID0+IGF3YWl0IFRhc2suRnJvbVJlc3VsdChuZXcgU2NyaXB0UmVzcG9uc2UgeyBEYXRhID0gY29udGV4dC5Cb2R5IH0pOwp9Cg==" + } + } + ] + } + ] + }, + { + "key": "approved", + "stateType": 3, + "subType": 1, + "versionStrategy": "Major", + "labels": [ + { + "language": "en-US", + "label": "Approved" + } + ], + "view": null, + "subFlow": null, + "onEntries": [], + "onExits": [], + "transitions": [] + } + ] + } +} \ No newline at end of file diff --git a/core/Workflows/event-driven-lab/src/ApproveEventMapping.csx b/core/Workflows/event-driven-lab/src/ApproveEventMapping.csx new file mode 100644 index 0000000..dacb858 --- /dev/null +++ b/core/Workflows/event-driven-lab/src/ApproveEventMapping.csx @@ -0,0 +1,28 @@ +using System.Threading.Tasks; +using BBT.Workflow.Scripting; + +/// +/// Transition-level event mapping for approve-by-event (triggerType: 3). +/// +/// Correlates by business key: the runtime looks up the ACTIVE instance whose key equals +/// InstanceKey. No match is a normal answer — the endpoint acks so the broker does not +/// redeliver forever — which is why the test asserts on the instance rather than on the response. +/// +/// +public class ApproveEventMapping : IEventMapping +{ + public async Task Handler(ScriptContext context) + { + var payload = context.EventPayload; + + return await Task.FromResult(new EventMappingResult + { + InstanceKey = (string)payload.orderId, + Body = new + { + decision = (string)payload.decision, + approvedBy = (string)payload.approvedBy + } + }); + } +} diff --git a/core/Workflows/event-driven-lab/src/RecordApprovalMapping.csx b/core/Workflows/event-driven-lab/src/RecordApprovalMapping.csx new file mode 100644 index 0000000..a21779e --- /dev/null +++ b/core/Workflows/event-driven-lab/src/RecordApprovalMapping.csx @@ -0,0 +1,25 @@ +using System.Collections.Generic; +using System.Dynamic; +using System.Threading.Tasks; +using BBT.Workflow.Definitions; +using BBT.Workflow.Scripting; + +/// +/// Writes what the event carried into instance data, so the test can assert that the transition ran +/// with the mapped body rather than merely that the state moved. +/// +public class RecordApprovalMapping : IMapping +{ + public async Task InputHandler(WorkflowTask task, ScriptContext context) + { + dynamic body = new ExpandoObject(); + body.approvals = 1; + body.decision = context.Body?.decision; + body.approvedBy = context.Body?.approvedBy; + + return await Task.FromResult(new ScriptResponse { Data = body }); + } + + public async Task OutputHandler(ScriptContext context) + => await Task.FromResult(new ScriptResponse { Data = context.Body }); +} diff --git a/core/Workflows/event-driven-lab/src/StartEventMapping.csx b/core/Workflows/event-driven-lab/src/StartEventMapping.csx new file mode 100644 index 0000000..151a530 --- /dev/null +++ b/core/Workflows/event-driven-lab/src/StartEventMapping.csx @@ -0,0 +1,36 @@ +using System.Threading.Tasks; +using BBT.Workflow.Scripting; + +/// +/// Workflow-level event mapping: turns a raw broker message into a new instance. +/// +/// Reached through POST .../instances/events?action=start, which a Dapr Subscription routes +/// to from the topic. The runtime has already unwrapped the CloudEvent envelope, so +/// context.EventPayload is the message the publisher actually sent. +/// +/// +/// InstanceKey is the correlation contract for everything that follows: the approve event +/// later finds this instance by exactly this value, so it must come from the payload and not be +/// generated here. +/// +/// +public class StartEventMapping : IEventMapping +{ + public async Task Handler(ScriptContext context) + { + var payload = context.EventPayload; + + return await Task.FromResult(new EventMappingResult + { + InstanceKey = (string)payload.orderId, + Body = new + { + orderId = (string)payload.orderId, + amount = payload.amount, + // Proves the instance was created by the event path rather than by an HTTP start. + startedBy = "event", + approvals = 0 + } + }); + } +} diff --git a/tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-approve.yaml b/tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-approve.yaml new file mode 100644 index 0000000..7331957 --- /dev/null +++ b/tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-approve.yaml @@ -0,0 +1,16 @@ +# Routes the approval topic to the same endpoint with action=transition. +# +# transitionKey is required for action=transition and must name a transition whose triggerType is 3; +# anything else is answered 400 NotAnEventTransition. sync=true makes the delivery block until the +# pipeline reaches a rest point, which is what lets a test assert on the instance immediately after +# the publish is acknowledged instead of polling for an indeterminate window. +apiVersion: dapr.io/v1alpha1 +kind: Subscription +metadata: + name: event-driven-lab-approve-subscription +spec: + topic: core.event-driven-lab.approve + route: "/api/v1/core/workflows/event-driven-lab/instances/events?action=transition&transitionKey=approve-by-event&sync=true" + pubsubname: vnext-pubsub +scopes: + - vnext-app diff --git a/tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-start.yaml b/tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-start.yaml new file mode 100644 index 0000000..150bab5 --- /dev/null +++ b/tests/Core.IntegrationTests/Infrastructure/DaprComponents/orchestration/subscription-event-driven-lab-start.yaml @@ -0,0 +1,15 @@ +# Routes the start topic to the runtime's event endpoint. +# +# Subscriptions are DOMAIN-OWNED: the runtime exposes the endpoint and the domain decides which +# topic reaches it. That is why this file lives with the example domain rather than in the runtime, +# and why the route carries the domain and workflow in its path. +apiVersion: dapr.io/v1alpha1 +kind: Subscription +metadata: + name: event-driven-lab-start-subscription +spec: + topic: core.event-driven-lab + route: /api/v1/core/workflows/event-driven-lab/instances/events?action=start + pubsubname: vnext-pubsub +scopes: + - vnext-app diff --git a/tests/Core.IntegrationTests/Tests/EventDrivenLab/EventDrivenLabTests.cs b/tests/Core.IntegrationTests/Tests/EventDrivenLab/EventDrivenLabTests.cs new file mode 100644 index 0000000..325a432 --- /dev/null +++ b/tests/Core.IntegrationTests/Tests/EventDrivenLab/EventDrivenLabTests.cs @@ -0,0 +1,167 @@ +using System.Net; +using System.Net.Http.Json; +using System.Text.Json; +using Core.IntegrationTests.Infrastructure; + +namespace Core.IntegrationTests.Tests.EventDrivenLab; + +/// +/// The event-driven entry point, end to end over a real broker. +/// +/// This is the first coverage the instances/events endpoint has ever had. It is reached only +/// through a Dapr Subscription, so a test that posted to the endpoint directly would prove the +/// mapping script and nothing about the wiring that actually delivers production traffic — the +/// topic, the subscription route, the CloudEvent unwrap, and the triggerType: 3 gate. +/// +/// +/// Both messages are published to the broker; nothing here calls the runtime's HTTP API to drive the +/// flow. The instance is created by an event and advanced by an event. +/// +/// +public class EventDrivenLabTests : IntegrationTestBase +{ + private const string Workflow = "event-driven-lab"; + private const string StartTopic = "core.event-driven-lab"; + private const string ApproveTopic = "core.event-driven-lab.approve"; + private const string PubSub = "vnext-pubsub"; + + private readonly HttpClient _dapr; + + public EventDrivenLabTests(VNextTestEnvironment environment) : base(environment) + { + // Published through the orchestration sidecar, because that is the sidecar whose + // subscriptions are under test. Its port is the runtime's Dapr HTTP port, recorded per + // domain in ai-docs/local-environments/.md. + var daprUrl = System.Environment.GetEnvironmentVariable("VNEXT_DAPR_HTTP_URL") + ?? "http://localhost:42110"; + _dapr = new HttpClient { BaseAddress = new Uri(daprUrl.TrimEnd('/') + "/") }; + } + + [SkippableFact] + public async Task AnEventStartsTheInstance_AndASecondEventAdvancesIt() + { + Skip.If(System.Environment.GetEnvironmentVariable("VNEXT_DAPR_HTTP_URL") is null + && System.Environment.GetEnvironmentVariable("VNEXT_BASE_URL") is null, + "needs a running stack with a reachable Dapr sidecar"); + + var orderId = $"evt-{Guid.NewGuid():N}"[..16]; + + // 1. Start by event. The workflow-level mapping turns the payload into a new instance whose + // key is the orderId — the correlation contract the second event depends on. + await PublishAsync(StartTopic, new { orderId, amount = 250 }); + + var instanceId = await WaitForInstanceAsync(orderId); + var afterStart = await GetStateAsync(instanceId); + Assert.Equal("awaiting-approval", afterStart.State); + + var attributes = await GetAttributesAsync(instanceId); + Assert.Equal("event", attributes.GetProperty("startedBy").GetString()); + Assert.Equal(orderId, attributes.GetProperty("orderId").GetString()); + + // 2. Advance by event. Correlation is by business key only — no instance id is published, + // which is the whole point of the mapping's InstanceKey. + await PublishAsync(ApproveTopic, new { orderId, decision = "approved", approvedBy = "ops" }); + + await WaitUntilAsync( + async () => (await GetStateAsync(instanceId)).State == "approved", + $"the approval event never advanced {instanceId}", + TimeSpan.FromSeconds(60)); + + // The transition ran with the MAPPED body, not merely with a state change: an event that + // moved the state but dropped its payload would pass a state-only assertion. + var final = await GetAttributesAsync(instanceId); + Assert.Equal("approved", final.GetProperty("decision").GetString()); + Assert.Equal("ops", final.GetProperty("approvedBy").GetString()); + } + + /// + /// An event naming no active instance is acked, not retried. The endpoint answers the broker's + /// protocol, so a "nothing matched" result must not look like a failure — otherwise the broker + /// redelivers a message this runtime can never satisfy. + /// + [SkippableFact] + public async Task AnApprovalForAnUnknownOrder_IsAcceptedAndChangesNothing() + { + Skip.If(System.Environment.GetEnvironmentVariable("VNEXT_DAPR_HTTP_URL") is null + && System.Environment.GetEnvironmentVariable("VNEXT_BASE_URL") is null, + "needs a running stack with a reachable Dapr sidecar"); + + var response = await PublishAsync( + ApproveTopic, + new { orderId = $"missing-{Guid.NewGuid():N}"[..20], decision = "approved", approvedBy = "ops" }); + + Assert.True( + response.IsSuccessStatusCode, + $"the broker publish was rejected with {response.StatusCode}"); + } + + private async Task PublishAsync(string topic, object payload) + { + var response = await _dapr.PostAsJsonAsync($"v1.0/publish/{PubSub}/{topic}", payload); + Assert.True( + response.IsSuccessStatusCode, + $"publish to {topic} failed with {response.StatusCode}"); + return response; + } + + /// + /// Finds the instance the start event created, by its business key. The publish returns as soon + /// as the broker accepts the message, so the instance appears asynchronously. + /// + private async Task WaitForInstanceAsync(string key) + { + string? instanceId = null; + + await WaitUntilAsync( + async () => + { + // Filtered by business key, because the publish returns no instance id — correlation + // by key is exactly what the mapping's InstanceKey establishes. + var response = await Api.ListInstancesAsync( + Workflow, + new Dictionary + { + ["filter"] = $"{{\"key\":{{\"eq\":\"{key}\"}}}}" + }); + if (!response.IsSuccessStatusCode) return false; + + if (!response.Body.TryGetProperty("items", out var items) || + items.GetArrayLength() == 0) + { + return false; + } + + instanceId = items[0].GetProperty("id").GetString(); + return instanceId is not null; + }, + $"no instance was created for key {key} — the start subscription never reached the runtime", + TimeSpan.FromSeconds(60)); + + return instanceId!; + } + + private async Task<(string State, string Status)> GetStateAsync(string instanceId) + { + var response = await Api.CallInstanceFunctionAsync(Workflow, instanceId, "state"); + return (response.Body.GetProperty("state").GetString() ?? "", + response.Body.GetProperty("status").GetString() ?? ""); + } + + private async Task GetAttributesAsync(string instanceId) + { + var response = await Api.GetInstanceAsync(Workflow, instanceId); + return response.Body.GetProperty("attributes"); + } + + private static async Task WaitUntilAsync(Func> condition, string because, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await condition()) return; + await Task.Delay(500); + } + + Assert.Fail($"Timed out after {timeout.TotalSeconds:0}s: {because}"); + } +} diff --git a/tests/Core.IntegrationTests/Tests/EventDrivenLab/README.md b/tests/Core.IntegrationTests/Tests/EventDrivenLab/README.md new file mode 100644 index 0000000..bee9e50 --- /dev/null +++ b/tests/Core.IntegrationTests/Tests/EventDrivenLab/README.md @@ -0,0 +1,59 @@ +# event-driven-lab + +Olay tetiklemeli akış: instance bir broker mesajıyla **doğar**, ikinci bir mesajla **ilerler**. +Testin hiçbir adımı akışı sürmek için runtime'ın HTTP API'sini çağırmaz. + +## Neden eklendi + +`POST .../instances/events` ucunun bu senaryodan önce **hiçbir entegrasyon testi yoktu** — agent +council `2026-09-11-route-trace-coverage` bunu adı konmuş bir boşluk olarak kaydetmişti. Uca +doğrudan POST atan bir test yalnız mapping script'ini kanıtlardı; üretim trafiğini gerçekten taşıyan +tesisatı (topic, subscription route'u, CloudEvent açma, `triggerType: 3` kapısı) kanıtlamazdı. +Bu yüzden iki mesaj da broker'a yayınlanır. + +İkincil amaç: `Event.Intake` span'i bu ucu ölçüyor ve "testi olmayan route'a span gönderilmez" +kuralı gereği span ile senaryo aynı commit'te gelir. + +## Yapılandırma + +Doküman: https://burgan-tech.github.io/vnext-docs/docs/how-to/event-driven-workflows + +| Parça | Yer | +| --- | --- | +| Akış-seviyesi `event.mapping` (action=start) | `core/Workflows/event-driven-lab/src/StartEventMapping.csx` | +| `triggerType: 3` geçiş + kendi mapping'i | `approve-by-event`, `src/ApproveEventMapping.csx` | +| Subscription'lar | `tests/.../DaprComponents/orchestration/subscription-event-driven-lab-*.yaml` | + +Topic'ler: `core.event-driven-lab` (start) ve `core.event-driven-lab.approve` (transition). +Approve subscription'ı `sync=true` taşır — teslimat pipeline dinlenme noktasına ulaşana kadar +bloklar, böylece test belirsiz bir pencere boyunca yoklamak yerine hemen doğrulayabilir. + +`InstanceKey` korelasyon sözleşmesidir: start mapping'i onu payload'daki `orderId`'den üretir, +approve mapping'i aynı değerle **aktif** instance'ı bulur. Yayınlanan mesajda instance id yoktur. + +## Neyi pinler + +1. Instance olayla doğar ve `startedBy = "event"` verisiyle `awaiting-approval`'da bekler. +2. İkinci olay onu `approved`'a taşır **ve mapping'lenmiş gövdeyi yazar** — yalnız state kontrol eden + bir assertion, payload'ı düşüren bir olayı yakalayamazdı. +3. Aktif instance'ı olmayan bir onay olayı **ack'lenir**, yeniden denenmez. Uç broker'ın protokolünü + yanıtlar; "eşleşme yok" bir hata gibi görünürse broker asla karşılanamayacak bir mesajı sonsuza + kadar yeniden teslim eder. + +## Çalıştırma + +Ayakta olan bir stack'e karşı: + +```bash +VNEXT_BASE_URL=http://localhost:4201 VNEXT_DAPR_HTTP_URL=http://localhost:42110 \ + dotnet test tests/Core.IntegrationTests --filter "FullyQualifiedName~EventDrivenLab" +``` + +`VNEXT_DAPR_HTTP_URL` orkestrasyon sidecar'ının Dapr HTTP portudur (domain başına +`ai-docs/local-environments/.md` içinde yazılı), çünkü abonelikleri test edilen sidecar odur. + +## Bilinen tuzak + +**Dapr declarative subscription'ları YALNIZ açılışta okur.** Subscription YAML'ı sidecar ayaktayken +eklerseniz mesaj sessizce hiçbir yere gitmez — hata da log da yok. `docker restart +vnext-orchestration-dapr` ve log'da `Found Subscription: …` satırını görün.