diff --git a/README.md b/README.md index 8ddf062..01b325e 100644 --- a/README.md +++ b/README.md @@ -52,13 +52,11 @@ codeunit 50100 "Exchange Rate Worker" implements "PW IParallel Worker" BaseCurrency: Text; CurrencyCode: Text; Rate: Decimal; - i: Integer; begin Items := Ctx.GetItems(); // currencies assigned to this chunk BaseCurrency := Ctx.GetTextInput('BaseCurrency'); // shared payload, same for every chunk - for i := 1 to Items.Count() do begin - Items.Get(i, Token); + foreach Token in Items do begin // JsonArray is 0-based — prefer foreach over index loops CurrencyCode := Token.AsValue().AsText(); Rate := FetchRateFromApi(BaseCurrency, CurrencyCode); // your HTTP call Rates.Add(CurrencyCode, Rate); @@ -86,7 +84,7 @@ enumextension 50100 "My Worker Types" extends "PW Worker Type" ### 3. Run it -`RunAndWaitForList` runs the batch, blocks until it finishes, collects results, and cleans up — one call for the common case. +`RunAndWaitForList` runs the batch, blocks until it finishes, and collects results — one call for the common case. Batch records are kept for monitoring by default; use the `AutoCleanup` overload or `Cleanup(BatchId)` to delete them. ```al procedure RefreshExchangeRates() @@ -267,13 +265,10 @@ procedure Execute(var Ctx: Codeunit "PW Chunk Context") var Items: JsonArray; Token: JsonToken; - i: Integer; begin Items := Ctx.GetItems(); - for i := 1 to Items.Count() do begin - Items.Get(i, Token); + foreach Token in Items do // JsonArray is 0-based — prefer foreach over index loops ProcessItem(Token.AsValue().AsText()); - end; end; // Caller @@ -327,26 +322,34 @@ var Coordinator: Codeunit "PW Batch Coordinator"; begin BatchId := Coordinator - .SetThreads(8) // Background sessions (default: 4) + .SetThreads(8) // Background sessions (default: 4, must be >= 1) .SetBatchTimeout(60) // Caller stops waiting after 60s — sessions keep running (default: 0 = wait forever) - .SetSessionTimeout(30000)// Platform kills any session after 30s — dead sessions auto-detected (default: 0 = no limit) + .SetSessionTimeout(30000)// Platform kills any session after 30s — dead sessions auto-detected (default: 0 = server default) .SetPollInterval(200) // Polling interval in ms (default: 500) .RunForList(...); ``` All four methods return `this`, so you can chain them. -### Simple API (run + wait + collect + cleanup in one call) +### Simple API (run + wait + collect in one call) | Method | Description | |---|---| -| `RunAndWaitForList(WorkerType, Items, var Results): Boolean` | Split a list, wait, collect results, clean up | +| `RunAndWaitForList(WorkerType, Items, var Results): Boolean` | Split a list, wait, collect results | | `RunAndWaitForList(WorkerType, Items, Payload, var Results): Boolean` | Same, with additional payload merged into each chunk | -| `RunAndWaitForRecords(WorkerType, RecRef, var Results): Boolean` | Split records, wait, collect results, clean up | +| `RunAndWaitForList(WorkerType, Items, Payload, AutoCleanup, var Results): Boolean` | Same, deletes batch records when done if `AutoCleanup` is true | +| `RunAndWaitForList(WorkerType, Items, Payload, AutoCleanup, var Results, var Errors): Boolean` | Same, also collects error messages from failed chunks | +| `RunAndWaitForRecords(WorkerType, RecRef, var Results): Boolean` | Split records, wait, collect results | | `RunAndWaitForRecords(WorkerType, RecRef, Payload, var Results): Boolean` | Same, with additional payload merged into each chunk | -| `RunAndWaitForChunks(WorkerType, Chunks, var Results): Boolean` | Run chunks, wait, collect results, clean up | +| `RunAndWaitForRecords(WorkerType, RecRef, Payload, AutoCleanup, var Results): Boolean` | Same, deletes batch records when done if `AutoCleanup` is true | +| `RunAndWaitForRecords(WorkerType, RecRef, Payload, AutoCleanup, var Results, var Errors): Boolean` | Same, also collects error messages from failed chunks | +| `RunAndWaitForChunks(WorkerType, Chunks, var Results): Boolean` | Run chunks, wait, collect results | +| `RunAndWaitForChunks(WorkerType, Chunks, AutoCleanup, var Results): Boolean` | Same, deletes batch records when done if `AutoCleanup` is true | +| `RunAndWaitForChunks(WorkerType, Chunks, AutoCleanup, var Results, var Errors): Boolean` | Same, also collects error messages from failed chunks | + +Returns `true` if all chunks succeeded, `false` on failure, partial failure, or timeout. On partial failure, `Results` still contains the output of the chunks that completed, and the `Errors` overloads collect the failure messages. Batch records are kept for monitoring unless `AutoCleanup` is `true` — use the granular API below if you need retry. -Returns `true` if all chunks succeeded. On failure, `Results` is empty — use the granular API below if you need error details or retry. +Empty input (empty list, empty record set, no chunks) is a no-op: the method returns `true` with empty `Results`. ### Granular API (for non-blocking, retry, progress polling) @@ -360,11 +363,13 @@ Returns `true` if all chunks succeeded. On failure, `Results` is empty — use t | `RunForRecords(WorkerType, RecRef, Payload): Guid` | Same, with additional payload merged into each chunk | | `RunForChunks(WorkerType, Chunks): Guid` | One chunk per JsonObject, no auto-splitting | +All `RunFor*` methods return a **null GUID** when there is nothing to process (empty list, empty record set, no chunks) — check with `IsNullGuid` before using the granular API. `WaitForCompletion` treats a null GUID as success. Payload keys prefixed with `$` are reserved for the framework and rejected with an error. + **Waiting & Status:** | Method | Description | |---|---| -| `WaitForCompletion(BatchId): Boolean` | Block until batch finishes. Returns `true` if all chunks succeeded. | +| `WaitForCompletion(BatchId): Boolean` | Block until batch finishes. Returns `true` if all chunks succeeded. Must not be called inside a write transaction — its recovery paths commit internally. | | `IsFinished(BatchId): Boolean` | Non-blocking check. | | `GetStatus(BatchId): Enum "PW Batch Status"` | Current status: `Running`, `Completed`, `PartialFailure`, `Failed`. | | `GetCompletedChunks(BatchId): Integer` | Number of successfully completed chunks. | @@ -382,7 +387,7 @@ Returns `true` if all chunks succeeded. On failure, `Results` is empty — use t | Method | Description | |---|---| -| `Cleanup(BatchId)` | Deletes the batch and all its chunk records. | +| `Cleanup(BatchId)` | Deletes the batch and all its chunk records. Safe to call while sessions are still running — their outcomes are simply discarded. | ## Chunk Context API @@ -441,7 +446,7 @@ The library enforces strict transaction boundaries: 2. **Workers cannot call Commit().** The dispatcher wraps your `Execute` call with `[CommitBehavior(CommitBehavior::Error)]`. If your worker tries to commit, a runtime error is raised and the chunk is marked as Failed. 3. **Every Commit is documented.** Every `Commit()` call has an inline comment explaining why it exists: - - Coordinator: 1 per `RunFor*` call (persist batch + chunks before `StartSession`). If any `StartSession` call fails, 2 additional commits handle failed chunk statuses and batch counter updates with `UpdLock`. + - Coordinator: 2 per `RunFor*` call (persist batch + chunks before `StartSession`; persist chunk session ids after all sessions are started). If any `StartSession` call fails, one more commit recounts batch counters with `UpdLock`. During `WaitForCompletion`, dead-session recovery and stalled-counter reconciliation commit when they repair state — which is why `WaitForCompletion` also refuses to run inside a write transaction. - Dispatcher: 3 per chunk at runtime (persist Running status, persist Completed/Failed status, release UpdLock after counter update) ```mermaid @@ -453,24 +458,26 @@ flowchart TD A2 -->|"No"| A4["INSERT Batch + Chunks"] A4 --> A5["COMMIT 1\nPersist records — StartSession\nfires immediately, sessions must\nfind chunk records on startup"] A5 --> A6["StartSession × N"] + A6 --> A7["COMMIT 2\nPersist chunk session ids —\ndead-session recovery needs them\nfor chunks that never reach Running"] end subgraph dispatcher["Dispatcher — background session, per chunk"] direction TB - B1["Mark chunk → Running"] --> B2["COMMIT 2\nRunning status visible to caller\npolling with ReadUncommitted"] + B1["Claim chunk (UpdLock)\nonly if still Pending\nMark chunk → Running"] --> B2["COMMIT 3\nRunning status visible to caller\npolling with ReadUncommitted"] B2 --> B3["Codeunit.Run WorkerRunner\nCommitBehavior::Error applied:\nworker Commit() → runtime error"] B3 -->|"Success"| B4["Save results, mark Completed"] B3 -->|"Error"| B5["Codeunit.Run rolls back all\nworker DB changes \nRe-read chunk, capture error\nMark Failed"] - B4 --> B6["COMMIT 3\nPersist chunk outcome\nbefore taking batch lock"] + B4 --> B6["COMMIT 4\nPersist chunk outcome\nbefore taking batch lock"] B5 --> B6 B6 --> B7["UpdLock on Batch row\nRecount chunk statuses\nSet terminal status if all done"] - B7 --> B8["COMMIT 4\nRelease lock, next dispatcher\nwaiting on UpdLock can proceed"] + B7 --> B8["COMMIT 5\nRelease lock, next dispatcher\nwaiting on UpdLock can proceed"] end A6 -.->|"starts"| B1 style A3 fill:#FEF2F2,stroke:#EF4444 style A5 fill:#DBEAFE,stroke:#3B82F6 + style A7 fill:#DBEAFE,stroke:#3B82F6 style B2 fill:#DBEAFE,stroke:#3B82F6 style B6 fill:#DBEAFE,stroke:#3B82F6 style B8 fill:#DBEAFE,stroke:#3B82F6 @@ -501,7 +508,13 @@ flowchart TD When `SetSessionTimeout` is configured, the platform kills background sessions that exceed the time limit. However, this is a hard kill — the dispatcher's error handling never runs, leaving the chunk stuck in `Running` status. -`WaitForCompletion` handles this automatically. After the session timeout plus a short margin, it checks each `Running` chunk using `Session.IsSessionActive`. If the session is dead, the chunk is marked `Failed` with a "Session terminated" error message, and batch counters are recounted using the same UpdLock pattern as `UpdateBatchCounters`. This ensures the batch reaches a terminal status (`Failed` or `PartialFailure`) even when sessions are killed by timeout. +`WaitForCompletion` handles this automatically. After the session timeout plus a short startup margin, it checks each unfinished chunk using `Session.IsSessionActive`, and keeps re-checking every couple of seconds — sessions start (and die) at different times, so a single check would miss late deaths. The coordinator records each chunk's session id when it starts the sessions, so even a `Pending` chunk whose session was killed before it could report `Running` is recovered. Dead chunks are marked `Failed` with a "Session terminated" error message, and batch counters are recounted using the same UpdLock pattern as `UpdateBatchCounters`. This ensures the batch reaches a terminal status (`Failed` or `PartialFailure`) even when sessions are killed by timeout. + +Note: `Session.IsSessionActive` is evaluated on the server instance running `WaitForCompletion`. In multi-node clusters, poll from the session that started the batch (the default blocking pattern does this) to avoid false positives. + +### Stalled Counter Reconciliation + +Independently of session timeouts, `WaitForCompletion` also repairs a rarer failure: a dispatcher that dies *between* committing its chunk's terminal status and updating the batch counters (for example, a lock timeout on the batch row). If every chunk has reached a committed terminal status while the batch row still says `Running` for two consecutive polls, the counters are recounted under `UpdLock` and the batch is moved to its terminal status — so the polling loop can never hang on a batch whose work is actually done. ### Timeout Behavior @@ -626,6 +639,7 @@ begin // Collect failed chunk inputs and retry them Coordinator.GetFailedChunkInputs(BatchId, FailedInputs); Coordinator.Cleanup(BatchId); + Commit(); // Cleanup leaves a write transaction open — RunFor* refuses to start inside one // Retry: use RunForChunks with the original payloads Clear(Chunks); @@ -676,6 +690,8 @@ The library ships with two pages for observing batch execution: `PW Batch Cleanup` (codeunit 99003) can be scheduled as a **Job Queue Entry**. By default, its `OnRun` trigger deletes all finished batches older than 24 hours. You can also call `CleanupOlderThan(Hours)` directly. +Only batches in a terminal status are deleted. Batches stuck in `Running` (for example, after a server restart killed their sessions) are never touched by the job — clean those up from the **PW Batches** page. + ## Worker Guidelines ### Do @@ -709,15 +725,28 @@ The library ships with two pages for observing batch execution: | Codeunit | 99004 | PW Worker Runner | Internal | | Page | 99000 | PW Batches | Public | | Page | 99001 | PW Batch Chunk List | Public | +| PermissionSet | 99000 | PW Parallel Worker | Assignable | +| Codeunit | 99100 | PW Sample Worker | Internal (sample) | +| Codeunit | 99101 | PW Sample Record Counter | Internal (sample) | +| Codeunit | 99102 | PW Sample Runner | Internal (sample) | +| Codeunit | 99103 | PW Sample Table Counter | Internal (sample) | +| Page | 99100 | PW Demo | Sample | +| EnumExt | 99100 | PW Sample Worker Type | Sample | ## Known Constraints -1. **No restart survival.** Background sessions created via `StartSession` don't survive server restarts. If the service restarts mid-batch, running chunks will be stuck. Use the monitoring pages to identify and clean up stuck batches. +1. **No restart survival.** Background sessions created via `StartSession` don't survive server restarts. If the service restarts mid-batch, running chunks will be stuck. The scheduled cleanup job only removes finished batches — use the monitoring pages to identify and clean up stuck ones. -2. **No cancellation.** Once a batch is started, there's no way to cancel running chunks. You can wait for them to finish, let `SetBatchTimeout` stop polling, or use `SetSessionTimeout` to have the platform kill sessions that exceed a time limit. +2. **No cancellation.** Once a batch is started, there's no way to cancel running chunks. You can wait for them to finish, let `SetBatchTimeout` stop polling, or use `SetSessionTimeout` to have the platform kill sessions that exceed a time limit. Calling `Cleanup` on a live batch is safe, but the running chunks' outcomes are discarded. 3. **No cross-chunk communication.** Each chunk is fully isolated. If you need map-reduce style aggregation, do the "reduce" step in your caller after `GetResults`. 4. **Session limits.** Business Central has per-tenant limits on concurrent background sessions. Don't set thread count too high — 4-8 is typically sufficient. -5. **Commit required before RunFor\*.** The coordinator must commit internally. You cannot have pending write transactions when calling any `RunFor*` method. \ No newline at end of file +5. **Commit required before RunFor\*.** The coordinator must commit internally. You cannot have pending write transactions when calling any `RunFor*` method — nor when calling `WaitForCompletion`, whose recovery paths also commit. + +6. **Stable record sets for RunForRecords.** Chunk ranges are positional (record index within the filtered set at batch creation time). If records matching the filter are inserted or deleted while the batch runs, affected workers fail with a "record range no longer matches the data" error instead of silently processing another chunk's records. Run record batches on data that is stable for the duration of the batch. + +7. **Reserved payload keys.** Payload keys prefixed with `$` (`$Items`, `$TableNo`, `$FilterView`, `$StartIndex`, `$EndIndex`, `$IsRecordChunk`) are reserved for the framework. `RunForList` and `RunForRecords` reject payloads containing them. + +8. **Batch timeout defaults to waiting forever.** With `SetBatchTimeout` unset (0), `WaitForCompletion` polls until the batch reaches a terminal status. The recovery mechanisms above make a permanently-`Running` batch very unlikely, but for defense in depth set an explicit batch timeout in unattended code. \ No newline at end of file diff --git a/app/API/PWBatchCoordinator.Codeunit.al b/app/API/PWBatchCoordinator.Codeunit.al index 637b01b..6272c5e 100644 --- a/app/API/PWBatchCoordinator.Codeunit.al +++ b/app/API/PWBatchCoordinator.Codeunit.al @@ -10,17 +10,23 @@ codeunit 99000 "PW Batch Coordinator" BatchTimeoutSeconds: Integer; SessionTimeoutMs: Integer; PollIntervalMs: Integer; - DeadSessionChecked: Boolean; + NextDeadSessionCheckAt: DateTime; + StalledBatchSeen: Boolean; WriteTransactionErr: Label 'Cannot start a parallel batch while a write transaction is active. Commit or rollback your changes before calling RunFor* methods, otherwise they would be silently committed.'; + WaitWriteTransactionErr: Label 'Cannot wait for a parallel batch while a write transaction is active. WaitForCompletion may commit internally during recovery, which would silently commit your changes.'; SessionTerminatedErr: Label 'Session terminated (timeout or server stop).'; + InvalidThreadCountErr: Label 'Thread count must be at least 1.'; + ReservedKeyErr: Label 'The payload key %1 is not allowed. Keys prefixed with $ are reserved for the framework.', Comment = '%1 = JSON key name'; /// /// Sets the number of parallel threads (background sessions) to use. /// - /// The number of threads. Defaults to 4 if not set. + /// The number of threads. Must be at least 1. Defaults to 4 if not set. /// The current instance for fluent chaining. procedure SetThreads(Count: Integer): Codeunit "PW Batch Coordinator" begin + if Count < 1 then + Error(InvalidThreadCountErr); ThreadCount := Count; exit(this); end; @@ -43,7 +49,7 @@ codeunit 99000 "PW Batch Coordinator" /// If a session exceeds this duration, the platform terminates it and the chunk is marked as Failed. /// This is a server-side timeout — protects against hung or runaway workers. /// - /// The timeout in milliseconds. Zero or unset means no limit. + /// The timeout in milliseconds. Zero or unset leaves the platform's default background session timeout in charge. /// The current instance for fluent chaining. procedure SetSessionTimeout(Milliseconds: Integer): Codeunit "PW Batch Coordinator" begin @@ -100,6 +106,7 @@ codeunit 99000 "PW Batch Coordinator" i: Integer; begin GuardNoWriteTransaction(); + AssertNoReservedKeys(BasePayload); TotalCount := RecRef.Count(); if TotalCount = 0 then @@ -180,6 +187,7 @@ codeunit 99000 "PW Batch Coordinator" j: Integer; begin GuardNoWriteTransaction(); + AssertNoReservedKeys(BasePayload); if Items.Count() = 0 then exit; @@ -254,6 +262,7 @@ codeunit 99000 "PW Batch Coordinator" /// /// The batch identifier returned by a RunFor* method. /// True if the batch completed successfully; false on failure, partial failure, or timeout. + /// Must not be called while a write transaction is active — recovery paths commit internally. procedure WaitForCompletion(BatchId: Guid): Boolean var Batch: Record "PW Batch"; @@ -263,6 +272,11 @@ codeunit 99000 "PW Batch Coordinator" if IsNullGuid(BatchId) then exit(true); + // Recovery below commits internally, which would silently persist any + // pending caller changes — same rule as the RunFor* methods. + if Database.IsInWriteTransaction() then + Error(WaitWriteTransactionErr); + // ReadUncommitted prevents polling from being blocked by // the dispatcher's UpdLock in UpdateBatchCounters. // Slightly stale data is acceptable — we retry every PollInterval. @@ -270,7 +284,8 @@ codeunit 99000 "PW Batch Coordinator" Interval := GetEffectivePollInterval(); StartTime := CurrentDateTime(); - DeadSessionChecked := false; + NextDeadSessionCheckAt := 0DT; + StalledBatchSeen := false; repeat if not Batch.Get(BatchId) then @@ -288,6 +303,8 @@ codeunit 99000 "PW Batch Coordinator" if ShouldCheckDeadSessions(StartTime) then RecoverDeadSessions(BatchId); + ReconcileStalledBatch(BatchId); + if (BatchTimeoutSeconds > 0) and ((CurrentDateTime() - StartTime) > BatchTimeoutSeconds * 1000) then exit(false); Sleep(Interval); @@ -392,7 +409,6 @@ codeunit 99000 "PW Batch Coordinator" var Chunk: Record "PW Batch Chunk"; InStream: InStream; - FullError: Text; begin Clear(Errors); Chunk.SetAutoCalcFields("Full Error Message"); @@ -402,8 +418,7 @@ codeunit 99000 "PW Batch Coordinator" repeat if Chunk."Full Error Message".HasValue() then begin Chunk."Full Error Message".CreateInStream(InStream, TextEncoding::UTF8); - InStream.ReadText(FullError); - Errors.Add(FullError); + Errors.Add(ReadBlobText(InStream)); end else if Chunk."Error Message" <> '' then Errors.Add(Chunk."Error Message"); @@ -657,14 +672,6 @@ codeunit 99000 "PW Batch Coordinator" exit(500); end; - local procedure GetEffectiveSessionTimeout(): Duration - begin - if SessionTimeoutMs > 0 then - exit(SessionTimeoutMs); - // 0 means no limit — max Duration effectively disables the timeout - exit(0); - end; - local procedure WriteFullErrorToChunk(var Chunk: Record "PW Batch Chunk"; ErrorText: Text) var OutStream: OutStream; @@ -673,14 +680,53 @@ codeunit 99000 "PW Batch Coordinator" OutStream.WriteText(ErrorText); end; + local procedure ReadBlobText(InStream: InStream): Text + var + Builder: TextBuilder; + Line: Text; + begin + // ReadText stops at line breaks — loop to preserve multi-line content. + while not InStream.EOS() do begin + InStream.ReadText(Line); + if Builder.Length() > 0 then + Builder.Append(NewLineText()); + Builder.Append(Line); + end; + exit(Builder.ToText()); + end; + + local procedure NewLineText(): Text + var + LineFeed: Char; + begin + LineFeed := 10; + exit(Format(LineFeed)); + end; + + local procedure AssertNoReservedKeys(BasePayload: JsonObject) + var + PayloadKey: Text; + begin + foreach PayloadKey in BasePayload.Keys() do + if PayloadKey.StartsWith('$') then + Error(ReservedKeyErr, PayloadKey); + end; + local procedure ShouldCheckDeadSessions(StartTime: DateTime): Boolean + var + ArmDelay: Duration; begin if SessionTimeoutMs <= 0 then exit(false); - if DeadSessionChecked then + // Arm after session timeout + 2s margin for session startup overhead. + // Duration arithmetic avoids Integer overflow near MaxInt timeouts. + ArmDelay := SessionTimeoutMs; + ArmDelay += 2000; + if (CurrentDateTime() - StartTime) <= ArmDelay then exit(false); - // Check after session timeout + 2s margin for session startup overhead - exit((CurrentDateTime() - StartTime) > (SessionTimeoutMs + 2000)); + // Sessions start — and die — at different times (queued sessions, + // staggered starts), so re-check periodically rather than once. + exit(CurrentDateTime() >= NextDeadSessionCheckAt); end; local procedure RecoverDeadSessions(BatchId: Guid) @@ -688,32 +734,80 @@ codeunit 99000 "PW Batch Coordinator" Chunk: Record "PW Batch Chunk"; AnyRecovered: Boolean; begin - DeadSessionChecked := true; + NextDeadSessionCheckAt := CurrentDateTime() + 2000; Chunk.SetRange("Batch Id", BatchId); - Chunk.SetRange(Status, "PW Chunk Status"::Running); + Chunk.SetFilter(Status, '%1|%2', "PW Chunk Status"::Pending, "PW Chunk Status"::Running); + // Pending chunks carry the session id stored by StartBatch. A session id + // of 0 means the session never started — the failed-to-start path in + // StartBatch already handles those chunks. + Chunk.SetFilter("Session Id", '<>0'); if Chunk.FindSet() then repeat - if not Session.IsSessionActive(Chunk."Session Id") then begin - Chunk."Error Message" := CopyStr(SessionTerminatedErr, 1, MaxStrLen(Chunk."Error Message")); - WriteFullErrorToChunk(Chunk, SessionTerminatedErr); - Chunk.Status := "PW Chunk Status"::Failed; - Chunk."Completed At" := CurrentDateTime(); - Chunk.Modify(); - AnyRecovered := true; - end; + if not Session.IsSessionActive(Chunk."Session Id") then + if MarkDeadChunkFailed(Chunk."Batch Id", Chunk."Chunk Index") then + AnyRecovered := true; until Chunk.Next() = 0; if AnyRecovered then begin + // Persist failed chunk statuses and release their row locks + // before taking the batch UpdLock in RecountBatchCounters. Commit(); RecountBatchCounters(BatchId); end; end; + local procedure MarkDeadChunkFailed(BatchId: Guid; ChunkIdx: Integer): Boolean + var + Chunk: Record "PW Batch Chunk"; + begin + // UpdLock + status re-check closes the race where the dispatcher finishes + // the chunk between the polling snapshot and this update — a plain Modify + // could otherwise fail the version check or overwrite a terminal status. + Chunk.ReadIsolation := IsolationLevel::UpdLock; + if not Chunk.Get(BatchId, ChunkIdx) then + exit(false); + if not (Chunk.Status in ["PW Chunk Status"::Pending, "PW Chunk Status"::Running]) then + exit(false); + + Chunk."Error Message" := CopyStr(SessionTerminatedErr, 1, MaxStrLen(Chunk."Error Message")); + WriteFullErrorToChunk(Chunk, SessionTerminatedErr); + Chunk.Status := "PW Chunk Status"::Failed; + Chunk."Completed At" := CurrentDateTime(); + Chunk.Modify(); + exit(true); + end; + + local procedure ReconcileStalledBatch(BatchId: Guid) + var + Chunk: Record "PW Batch Chunk"; + begin + // A dispatcher that dies between committing its chunk status and updating + // the batch counters leaves the batch Running forever. If every chunk has + // a committed terminal status while the batch still says Running, recount. + // Requiring the condition on two consecutive polls avoids grabbing the + // batch UpdLock while the last dispatcher is about to update it anyway. + Chunk.ReadIsolation := IsolationLevel::ReadCommitted; + Chunk.SetRange("Batch Id", BatchId); + Chunk.SetFilter(Status, '%1|%2', "PW Chunk Status"::Pending, "PW Chunk Status"::Running); + if not Chunk.IsEmpty() then begin + StalledBatchSeen := false; + exit; + end; + + if not StalledBatchSeen then begin + StalledBatchSeen := true; + exit; + end; + + StalledBatchSeen := false; + RecountBatchCounters(BatchId); + end; + /// /// Recounts chunk statuses and updates batch counters with UpdLock serialization. - /// Shared by StartBatch (failed-to-start recovery) and RecoverDeadSessions. - /// Same pattern as UpdateBatchCounters in PW Task Dispatcher. + /// Shared by StartBatch (failed-to-start recovery), RecoverDeadSessions and + /// ReconcileStalledBatch. Same pattern as UpdateBatchCounters in PW Task Dispatcher. /// local procedure RecountBatchCounters(BatchId: Guid) var @@ -723,7 +817,9 @@ codeunit 99000 "PW Batch Coordinator" FailedCount: Integer; begin Batch.ReadIsolation := IsolationLevel::UpdLock; - Batch.Get(BatchId); + // The batch disappears if the caller runs Cleanup mid-batch. + if not Batch.Get(BatchId) then + exit; Chunk.ReadIsolation := IsolationLevel::ReadCommitted; Chunk.SetRange("Batch Id", BatchId); @@ -785,7 +881,9 @@ codeunit 99000 "PW Batch Coordinator" var Batch: Record "PW Batch"; Chunk: Record "PW Batch Chunk"; + StartedSessions: Dictionary of [Integer, Integer]; SessionId: Integer; + ChunkIdx: Integer; FailedToStart: Integer; begin Batch.Get(BatchId); @@ -793,7 +891,9 @@ codeunit 99000 "PW Batch Coordinator" Chunk.SetRange("Batch Id", BatchId); if Chunk.FindSet() then repeat - if not StartSession(SessionId, Codeunit::"PW Task Dispatcher", Batch."Company Name", Chunk, GetEffectiveSessionTimeout()) then begin + if StartDispatcherSession(SessionId, Batch."Company Name", Chunk) then + StartedSessions.Add(Chunk."Chunk Index", SessionId) + else begin Chunk."Error Message" := CopyStr(GetLastErrorText(), 1, MaxStrLen(Chunk."Error Message")); WriteFullErrorToChunk(Chunk, GetLastErrorText()); Chunk.Status := "PW Chunk Status"::Failed; @@ -803,13 +903,44 @@ codeunit 99000 "PW Batch Coordinator" end; until Chunk.Next() = 0; - if FailedToStart > 0 then begin - // Commit failed-to-start chunk statuses first, so background sessions - // that are already running can see them in UpdateBatchCounters. - // This also releases all locks from the FindSet loop, preventing - // deadlocks when we take UpdLock below. - Commit(); + // Session ids are stored after all sessions are started: storing inside + // the loop would hold each chunk's row lock for the whole StartSession + // sequence and lock-timeout dispatchers on their first read. + foreach ChunkIdx in StartedSessions.Keys() do + StoreSessionId(BatchId, ChunkIdx, StartedSessions.Get(ChunkIdx)); + + // Persist session ids (dead-session recovery needs them for chunks that + // never reach Running) and failed-to-start statuses, release all row + // locks, and leave the caller without an open write transaction — + // RunFor* must return with a clean transaction state. + Commit(); + + if FailedToStart > 0 then RecountBatchCounters(BatchId); - end; + end; + + local procedure StartDispatcherSession(var NewSessionId: Integer; RunInCompany: Text; var Chunk: Record "PW Batch Chunk"): Boolean + begin + // An explicit 0 for the timeout parameter is undocumented platform + // behavior — when no session timeout is configured, use the overload + // that leaves the server default in charge. + if SessionTimeoutMs > 0 then + exit(StartSession(NewSessionId, Codeunit::"PW Task Dispatcher", RunInCompany, Chunk, SessionTimeoutMs)); + exit(StartSession(NewSessionId, Codeunit::"PW Task Dispatcher", RunInCompany, Chunk)); + end; + + local procedure StoreSessionId(BatchId: Guid; ChunkIdx: Integer; NewSessionId: Integer) + var + Chunk: Record "PW Batch Chunk"; + begin + // UpdLock serializes with the dispatcher claiming this chunk. If the + // dispatcher won the race, the chunk is no longer Pending and already + // carries its own session id. + Chunk.ReadIsolation := IsolationLevel::UpdLock; + Chunk.Get(BatchId, ChunkIdx); + if Chunk.Status <> "PW Chunk Status"::Pending then + exit; + Chunk."Session Id" := NewSessionId; + Chunk.Modify(); end; } diff --git a/app/API/PWChunkContext.Codeunit.al b/app/API/PWChunkContext.Codeunit.al index 7174a7f..1a03277 100644 --- a/app/API/PWChunkContext.Codeunit.al +++ b/app/API/PWChunkContext.Codeunit.al @@ -14,6 +14,7 @@ codeunit 99001 "PW Chunk Context" KeyNotFoundErr: Label 'Key "%1" not found in chunk input.', Comment = '%1 = JSON key name'; NotRecordChunkErr: Label 'This chunk was not created via RunForRecords.'; NotInitializedErr: Label 'Chunk context has not been initialized.'; + RecordSetChangedErr: Label 'The chunk''s record range no longer matches the data: the filtered record set is empty or has shrunk since the batch was created.'; procedure Init(var Chunk: Record "PW Batch Chunk") var @@ -202,9 +203,14 @@ codeunit 99001 "PW Chunk Context" if FilterView <> '' then RecRef.SetView(FilterView); - RecRef.FindSet(); + // Chunk ranges are positional — if records were inserted or deleted + // since the batch was created, the ranges no longer match the data. + // Fail loudly instead of silently processing another chunk's records. + if not RecRef.FindSet() then + Error(RecordSetChangedErr); if StartIndex > 1 then - RecRef.Next(StartIndex - 1); + if RecRef.Next(StartIndex - 1) <> StartIndex - 1 then + Error(RecordSetChangedErr); end; /// @@ -222,6 +228,7 @@ codeunit 99001 "PW Chunk Context" /// The JSON object to store as the sole result. procedure SetResult(Result: JsonObject) begin + CheckInitialized(); Clear(Results); Results.Add(Result); end; @@ -232,6 +239,7 @@ codeunit 99001 "PW Chunk Context" /// The JSON object to append. procedure AppendResult(Result: JsonObject) begin + CheckInitialized(); Results.Add(Result); end; diff --git a/app/Engine/PWTaskDispatcher.Codeunit.al b/app/Engine/PWTaskDispatcher.Codeunit.al index 064ac08..e70095a 100644 --- a/app/Engine/PWTaskDispatcher.Codeunit.al +++ b/app/Engine/PWTaskDispatcher.Codeunit.al @@ -8,8 +8,16 @@ codeunit 99002 "PW Task Dispatcher" ErrorText: Text; ErrorCallStack: Text; begin + // UpdLock serializes the claim with the coordinator storing this chunk's + // session id in StartBatch and with dead-session recovery — a plain read + // could hit an optimistic-concurrency error on the Modify below. + Rec.ReadIsolation := IsolationLevel::UpdLock; if not Rec.Get(Rec."Batch Id", Rec."Chunk Index") then exit; + // Dead-session recovery may have already failed a chunk whose session + // was queued for too long — never resurrect a chunk that left Pending. + if Rec.Status <> "PW Chunk Status"::Pending then + exit; Rec.Status := "PW Chunk Status"::Running; Rec."Session Id" := SessionId(); @@ -29,21 +37,26 @@ codeunit 99002 "PW Task Dispatcher" if Codeunit.Run(Codeunit::"PW Worker Runner", Rec) then begin // Worker succeeded — results already saved by PW Worker Runner. // Re-read to pick up changes made inside Codeunit.Run. - Rec.Get(Rec."Batch Id", Rec."Chunk Index"); + // The chunk disappears if the caller ran Cleanup mid-batch — + // nothing left to persist. + if not Rec.Get(Rec."Batch Id", Rec."Chunk Index") then + exit; Rec.Status := "PW Chunk Status"::Completed; Rec."Completed At" := CurrentDateTime(); Rec.Modify(); // Persist chunk's Completed status before updating batch counters. // If UpdateBatchCounters fails (e.g. lock timeout on PW Batch), // the chunk is still correctly marked as Completed. - // The next finishing chunk will recompute counters correctly. + // WaitForCompletion reconciles the counters if no chunk finishes after this one. Commit(); end else begin // Worker failed — Codeunit.Run rolled back all DB changes. - // Re-read to ensure Rec is clean, consistent with the success branch. - Rec.Get(Rec."Batch Id", Rec."Chunk Index"); + // Capture the error before any further record operations, + // then re-read to ensure Rec is clean, consistent with the success branch. ErrorText := GetLastErrorText(); ErrorCallStack := GetLastErrorCallStack(); + if not Rec.Get(Rec."Batch Id", Rec."Chunk Index") then + exit; Rec."Error Message" := CopyStr(ErrorText, 1, MaxStrLen(Rec."Error Message")); WriteFullErrorMessage(Rec, ErrorText); WriteErrorCallStack(Rec, ErrorCallStack); @@ -83,7 +96,9 @@ codeunit 99002 "PW Task Dispatcher" // UpdLock on the Batch row serializes concurrent counter updates // without disabling Tri-State Locking for the entire table. Batch.ReadIsolation := IsolationLevel::UpdLock; - Batch.Get(BatchId); + // The batch disappears if the caller ran Cleanup mid-batch. + if not Batch.Get(BatchId) then + exit; // ReadCommitted ensures we see committed chunk statuses only. // Each dispatcher commits its chunk status before calling this procedure. diff --git a/app/Monitoring/PWBatchCleanup.Codeunit.al b/app/Monitoring/PWBatchCleanup.Codeunit.al index cf41b11..0d5315a 100644 --- a/app/Monitoring/PWBatchCleanup.Codeunit.al +++ b/app/Monitoring/PWBatchCleanup.Codeunit.al @@ -10,21 +10,18 @@ codeunit 99003 "PW Batch Cleanup" procedure CleanupOlderThan(Hours: Integer) var Batch: Record "PW Batch"; - Chunk: Record "PW Batch Chunk"; + CutoffOffset: Duration; begin + // Duration arithmetic — Hours * 3600000 as Integer overflows at ~596 hours. + CutoffOffset := Hours; + CutoffOffset := CutoffOffset * 3600000; + Batch.SetFilter(Status, '%1|%2|%3', "PW Batch Status"::Completed, "PW Batch Status"::PartialFailure, "PW Batch Status"::Failed); - Batch.SetFilter("Completed At", '<>%1', 0DT); - - if Batch.FindSet() then - repeat - if (CurrentDateTime() - Batch."Completed At") > Hours * 3600000 then begin - Chunk.SetRange("Batch Id", Batch.Id); - Chunk.DeleteAll(); - Batch.Delete(); - end; - until Batch.Next() = 0; + Batch.SetFilter("Completed At", '<>%1&<%2', 0DT, CurrentDateTime() - CutoffOffset); + // OnDelete on PW Batch removes the chunks. + Batch.DeleteAll(true); end; } diff --git a/app/Translations/Parallel Worker.g.xlf b/app/Translations/Parallel Worker.g.xlf index fd9ce2d..d4fd456 100644 --- a/app/Translations/Parallel Worker.g.xlf +++ b/app/Translations/Parallel Worker.g.xlf @@ -118,11 +118,26 @@ Table PW Batch Chunk - Field Status - Property Caption + + Thread count must be at least 1. + + Codeunit PW Batch Coordinator - NamedType InvalidThreadCountErr + + + The payload key %1 is not allowed. Keys prefixed with $ are reserved for the framework. + %1 = JSON key name + Codeunit PW Batch Coordinator - NamedType ReservedKeyErr + Session terminated (timeout or server stop). Codeunit PW Batch Coordinator - NamedType SessionTerminatedErr + + Cannot wait for a parallel batch while a write transaction is active. WaitForCompletion may commit internally during recovery, which would silently commit your changes. + + Codeunit PW Batch Coordinator - NamedType WaitWriteTransactionErr + Cannot start a parallel batch while a write transaction is active. Commit or rollback your changes before calling RunFor* methods, otherwise they would be silently committed. @@ -143,6 +158,11 @@ Codeunit PW Chunk Context - NamedType NotRecordChunkErr + + The chunk's record range no longer matches the data: the filtered record set is empty or has shrunk since the batch was created. + + Codeunit PW Chunk Context - NamedType RecordSetChangedErr + Chunk %1: %1 = chunk index diff --git a/app/Volodymyr Dvernytskyi_Parallel Worker_1.0.0.1.app b/app/Volodymyr Dvernytskyi_Parallel Worker_1.0.0.1.app index 1982234..b51112b 100644 Binary files a/app/Volodymyr Dvernytskyi_Parallel Worker_1.0.0.1.app and b/app/Volodymyr Dvernytskyi_Parallel Worker_1.0.0.1.app differ diff --git a/test/Volodymyr Dvernytskyi_Parallel Worker Test_1.0.0.0.app b/test/Volodymyr Dvernytskyi_Parallel Worker Test_1.0.0.0.app index 6bccb19..a77d66d 100644 Binary files a/test/Volodymyr Dvernytskyi_Parallel Worker Test_1.0.0.0.app and b/test/Volodymyr Dvernytskyi_Parallel Worker Test_1.0.0.0.app differ diff --git a/test/disabledtests.json b/test/disabledtests.json index a8574f9..18acf84 100644 --- a/test/disabledtests.json +++ b/test/disabledtests.json @@ -202,5 +202,17 @@ "codeunitId": 99209, "codeunitName": "PW Integration Test", "method": "BatchTimeoutReturnsFalseButSessionContinues" + }, + { + "bug": "Need to set isolation disabled test runner. Currently not supported by AL-Go.", + "codeunitId": 99209, + "codeunitName": "PW Integration Test", + "method": "SessionIdStoredOnChunksAtStart" + }, + { + "bug": "Need to set isolation disabled test runner. Currently not supported by AL-Go.", + "codeunitId": 99209, + "codeunitName": "PW Integration Test", + "method": "CleanupWhileSessionsRunningIsSafe" } ] diff --git a/test/src/Helpers/PWTestHelper.Codeunit.al b/test/src/Helpers/PWTestHelper.Codeunit.al index 1f4e50d..754869d 100644 --- a/test/src/Helpers/PWTestHelper.Codeunit.al +++ b/test/src/Helpers/PWTestHelper.Codeunit.al @@ -97,4 +97,31 @@ codeunit 99204 "PW Test Helper" Chunk.Status := Status; Chunk.Insert(); end; + + procedure CreateChunkWithSession(BatchId: Guid; ChunkIndex: Integer; Status: Enum "PW Chunk Status"; SessionId: Integer) + var + Chunk: Record "PW Batch Chunk"; + begin + Chunk."Batch Id" := BatchId; + Chunk."Chunk Index" := ChunkIndex; + Chunk.Status := Status; + Chunk."Session Id" := SessionId; + Chunk.Insert(); + end; + + procedure CreateChunkWithFullError(BatchId: Guid; ChunkIndex: Integer; ErrorText: Text) + var + Chunk: Record "PW Batch Chunk"; + OutStream: OutStream; + begin + Chunk."Batch Id" := BatchId; + Chunk."Chunk Index" := ChunkIndex; + Chunk.Status := "PW Chunk Status"::Failed; + Chunk."Error Message" := CopyStr(ErrorText, 1, MaxStrLen(Chunk."Error Message")); + Chunk.Insert(); + + Chunk."Full Error Message".CreateOutStream(OutStream, TextEncoding::UTF8); + OutStream.WriteText(ErrorText); + Chunk.Modify(); + end; } diff --git a/test/src/Tests/PWBatchCleanupTest.Codeunit.al b/test/src/Tests/PWBatchCleanupTest.Codeunit.al index e1a8cb7..59f6088 100644 --- a/test/src/Tests/PWBatchCleanupTest.Codeunit.al +++ b/test/src/Tests/PWBatchCleanupTest.Codeunit.al @@ -126,6 +126,24 @@ codeunit 99203 "PW Batch Cleanup Test" LibraryAssert.IsFalse(Batch.Get(BatchId), 'OnRun should delete old completed batch'); end; + [Test] + procedure CleanupOlderThanLargeHoursDoesNotOverflow() + // [SCENARIO 5.9] Hours large enough to overflow 32-bit millisecond math + // (Hours * 3600000 > MaxInt at ~596 hours) is handled without a runtime error + var + Cleanup: Codeunit "PW Batch Cleanup"; + Batch: Record "PW Batch"; + BatchId: Guid; + begin + BatchId := TestHelper.CreateBatchWithTimestamp( + "PW Batch Status"::Completed, + CurrentDateTime() - (48 * 3600000)); + + Cleanup.CleanupOlderThan(100000); + + LibraryAssert.IsTrue(Batch.Get(BatchId), '48h-old batch should survive a 100000h threshold'); + end; + [Test] procedure MixedScenarioDeletesOnlyOldFinished() // [SCENARIO 5.7] Mixed — only old finished batches are deleted diff --git a/test/src/Tests/PWChunkContextTest.Codeunit.al b/test/src/Tests/PWChunkContextTest.Codeunit.al index 7219f7d..2d3bcb8 100644 --- a/test/src/Tests/PWChunkContextTest.Codeunit.al +++ b/test/src/Tests/PWChunkContextTest.Codeunit.al @@ -507,4 +507,106 @@ codeunit 99201 "PW Chunk Context Test" Chunk.CalcFields("Result Payload"); LibraryAssert.IsFalse(Chunk."Result Payload".HasValue(), 'Result blob should be empty'); end; + + [Test] + procedure SetResultErrorsBeforeInit() + // [SCENARIO 1.23] SetResult called before Init raises error + var + Ctx: Codeunit "PW Chunk Context"; + Result: JsonObject; + begin + asserterror Ctx.SetResult(Result); + + LibraryAssert.ExpectedError('Chunk context has not been initialized.'); + end; + + [Test] + procedure AppendResultErrorsBeforeInit() + // [SCENARIO 1.24] AppendResult called before Init raises error + var + Ctx: Codeunit "PW Chunk Context"; + Result: JsonObject; + begin + asserterror Ctx.AppendResult(Result); + + LibraryAssert.ExpectedError('Chunk context has not been initialized.'); + end; + + [Test] + procedure GetRecordRefErrorsWhenRecordSetIsEmpty() + // [SCENARIO 1.25] GetRecordRef fails loudly when the chunk's filtered set is now empty + var + Ctx: Codeunit "PW Chunk Context"; + Chunk: Record "PW Batch Chunk"; + SourceBatch: Record "PW Batch"; + SourceRef: RecordRef; + TargetRef: RecordRef; + BatchId: Guid; + Payload: JsonObject; + begin + SourceBatch.SetRange("Company Name", 'PW_CTX_EMPTY'); + SourceRef.GetTable(SourceBatch); + + Payload.Add('$IsRecordChunk', true); + Payload.Add('$TableNo', Database::"PW Batch"); + Payload.Add('$FilterView', SourceRef.GetView()); + Payload.Add('$StartIndex', 1); + Payload.Add('$EndIndex', 5); + BatchId := TestHelper.CreateBatch("PW Batch Status"::Running); + TestHelper.CreateChunkWithPayload(BatchId, 1, Payload); + Chunk.Get(BatchId, 1); + Ctx.Init(Chunk); + + asserterror Ctx.GetRecordRef(TargetRef); + + LibraryAssert.ExpectedError('no longer matches the data'); + end; + + [Test] + procedure GetRecordRefErrorsWhenRecordSetShrank() + // [SCENARIO 1.26] GetRecordRef fails loudly when the set has fewer records + // than the chunk's start index — positioning would silently overlap another chunk + var + Ctx: Codeunit "PW Chunk Context"; + Chunk: Record "PW Batch Chunk"; + SourceBatch: Record "PW Batch"; + SourceRef: RecordRef; + TargetRef: RecordRef; + BatchId: Guid; + Payload: JsonObject; + i: Integer; + begin + SourceBatch.SetRange("Company Name", 'PW_CTX_SHRUNK'); + SourceBatch.DeleteAll(); + for i := 1 to 2 do begin + Clear(SourceBatch); + SourceBatch.Id := CreateGuid(); + SourceBatch.Status := "PW Batch Status"::Pending; + SourceBatch."Company Name" := 'PW_CTX_SHRUNK'; + SourceBatch.Insert(); + end; + + SourceBatch.Reset(); + SourceBatch.SetRange("Company Name", 'PW_CTX_SHRUNK'); + SourceRef.GetTable(SourceBatch); + + // Chunk expects to start at record 5 but only 2 records exist + Payload.Add('$IsRecordChunk', true); + Payload.Add('$TableNo', Database::"PW Batch"); + Payload.Add('$FilterView', SourceRef.GetView()); + Payload.Add('$StartIndex', 5); + Payload.Add('$EndIndex', 6); + BatchId := TestHelper.CreateBatch("PW Batch Status"::Running); + TestHelper.CreateChunkWithPayload(BatchId, 1, Payload); + Chunk.Get(BatchId, 1); + Ctx.Init(Chunk); + + asserterror Ctx.GetRecordRef(TargetRef); + + LibraryAssert.ExpectedError('no longer matches the data'); + + SourceBatch.Reset(); + SourceBatch.SetRange("Company Name", 'PW_CTX_SHRUNK'); + SourceBatch.DeleteAll(); + end; } diff --git a/test/src/Tests/PWCoordinatorTest.Codeunit.al b/test/src/Tests/PWCoordinatorTest.Codeunit.al index bea604c..af39c06 100644 --- a/test/src/Tests/PWCoordinatorTest.Codeunit.al +++ b/test/src/Tests/PWCoordinatorTest.Codeunit.al @@ -177,6 +177,7 @@ codeunit 99202 "PW Coordinator Test" BatchId: Guid; begin BatchId := TestHelper.CreateBatch("PW Batch Status"::Completed); + Commit(); // WaitForCompletion requires a clean transaction state LibraryAssert.IsTrue(Coordinator.WaitForCompletion(BatchId), 'Should return true for completed batch'); end; @@ -189,6 +190,7 @@ codeunit 99202 "PW Coordinator Test" BatchId: Guid; begin BatchId := TestHelper.CreateBatch("PW Batch Status"::Failed); + Commit(); // WaitForCompletion requires a clean transaction state LibraryAssert.IsFalse(Coordinator.WaitForCompletion(BatchId), 'Should return false for failed batch'); end; @@ -469,6 +471,25 @@ codeunit 99202 "PW Coordinator Test" LibraryAssert.AreEqual(1, FailedInputs.Count(), 'Should return 1 failed input'); end; + [Test] + procedure GetErrorsPreservesMultiLineFullError() + // [SCENARIO 3.13] GetErrors returns the full multi-line error, not just the first line + var + Coordinator: Codeunit "PW Batch Coordinator"; + BatchId: Guid; + Errors: List of [Text]; + MultiLineError: Text; + begin + MultiLineError := 'First line' + NewLine() + 'Second line' + NewLine() + 'Third line'; + BatchId := TestHelper.CreateBatch("PW Batch Status"::Failed); + TestHelper.CreateChunkWithFullError(BatchId, 1, MultiLineError); + + Coordinator.GetErrors(BatchId, Errors); + + LibraryAssert.AreEqual(1, Errors.Count(), 'Should return 1 error'); + LibraryAssert.AreEqual(MultiLineError, Errors.Get(1), 'Multi-line error text should be preserved'); + end; + [Test] procedure GetFailedChunkInputsReturnsEmptyForNoFailedChunks() // [SCENARIO 3.12] GetFailedChunkInputs — no failed chunks returns empty list @@ -550,4 +571,170 @@ codeunit 99202 "PW Coordinator Test" end; #endregion + + #region Recovery & Liveness + + [Test] + procedure WaitForCompletionReconcilesStalledCounters() + // [SCENARIO 14.1] All chunks terminal but batch counters stale (dispatcher died + // between its chunk commit and the counter update) — WaitForCompletion recounts + // and reaches a terminal status instead of polling forever + var + Coordinator: Codeunit "PW Batch Coordinator"; + BatchId: Guid; + begin + BatchId := TestHelper.CreateBatchEx("PW Batch Status"::Running, 2, 1, 0); + TestHelper.CreateSimpleChunk(BatchId, 1, "PW Chunk Status"::Completed); + TestHelper.CreateSimpleChunk(BatchId, 2, "PW Chunk Status"::Completed); + Commit(); // WaitForCompletion requires a clean transaction state + + LibraryAssert.IsTrue( + Coordinator.SetPollInterval(50).SetBatchTimeout(10).WaitForCompletion(BatchId), + 'Stalled batch should be reconciled to Completed'); + LibraryAssert.IsTrue( + Coordinator.GetStatus(BatchId) = "PW Batch Status"::Completed, + 'Status should be Completed after reconciliation'); + LibraryAssert.AreEqual(2, Coordinator.GetCompletedChunks(BatchId), 'Counters should be recounted'); + + Coordinator.Cleanup(BatchId); + Commit(); + end; + + [Test] + procedure WaitForCompletionRecoversDeadRunningChunk() + // [SCENARIO 14.2] Running chunk whose session no longer exists is marked Failed + // once the session timeout plus margin has elapsed + var + Coordinator: Codeunit "PW Batch Coordinator"; + Errors: List of [Text]; + BatchId: Guid; + begin + BatchId := TestHelper.CreateBatchEx("PW Batch Status"::Running, 1, 0, 0); + TestHelper.CreateChunkWithSession(BatchId, 1, "PW Chunk Status"::Running, 99999999); + Commit(); // WaitForCompletion requires a clean transaction state + + LibraryAssert.IsFalse( + Coordinator.SetSessionTimeout(1000).SetPollInterval(100).SetBatchTimeout(15).WaitForCompletion(BatchId), + 'Batch with a dead running session should fail'); + LibraryAssert.IsTrue( + Coordinator.GetStatus(BatchId) = "PW Batch Status"::Failed, + 'Status should be Failed'); + + Coordinator.GetErrors(BatchId, Errors); + LibraryAssert.AreEqual(1, Errors.Count(), 'Should have 1 error from the dead session'); + + Coordinator.Cleanup(BatchId); + Commit(); + end; + + [Test] + procedure WaitForCompletionRecoversDeadPendingChunk() + // [SCENARIO 14.3] Pending chunk with a stored session id whose session died + // before reaching Running is also recovered — not only Running chunks + var + Coordinator: Codeunit "PW Batch Coordinator"; + Errors: List of [Text]; + BatchId: Guid; + begin + BatchId := TestHelper.CreateBatchEx("PW Batch Status"::Running, 1, 0, 0); + TestHelper.CreateChunkWithSession(BatchId, 1, "PW Chunk Status"::Pending, 99999999); + Commit(); // WaitForCompletion requires a clean transaction state + + LibraryAssert.IsFalse( + Coordinator.SetSessionTimeout(1000).SetPollInterval(100).SetBatchTimeout(15).WaitForCompletion(BatchId), + 'Batch with a dead pending session should fail'); + LibraryAssert.IsTrue( + Coordinator.GetStatus(BatchId) = "PW Batch Status"::Failed, + 'Status should be Failed'); + + Coordinator.GetErrors(BatchId, Errors); + LibraryAssert.AreEqual(1, Errors.Count(), 'Should have 1 error from the dead session'); + + Coordinator.Cleanup(BatchId); + Commit(); + end; + + [Test] + procedure WaitForCompletionErrorsInsideWriteTransaction() + // [SCENARIO 14.4] WaitForCompletion inside a write transaction errors — + // its recovery paths commit internally + var + Coordinator: Codeunit "PW Batch Coordinator"; + BatchId: Guid; + begin + BatchId := TestHelper.CreateBatch("PW Batch Status"::Running); + + asserterror Coordinator.WaitForCompletion(BatchId); + + LibraryAssert.ExpectedError('Cannot wait for a parallel batch while a write transaction is active'); + end; + + #endregion + + #region Configuration Validation + + [Test] + procedure SetThreadsRejectsZero() + // [SCENARIO 15.1] SetThreads with zero errors instead of silently falling back + var + Coordinator: Codeunit "PW Batch Coordinator"; + begin + asserterror Coordinator.SetThreads(0); + + LibraryAssert.ExpectedError('Thread count must be at least 1'); + end; + + [Test] + procedure SetThreadsRejectsNegativeCount() + // [SCENARIO 15.2] SetThreads with a negative count errors + var + Coordinator: Codeunit "PW Batch Coordinator"; + begin + asserterror Coordinator.SetThreads(-3); + + LibraryAssert.ExpectedError('Thread count must be at least 1'); + end; + + [Test] + procedure RunForListRejectsReservedPayloadKey() + // [SCENARIO 15.3] RunForList rejects a payload containing a $-prefixed key + var + Coordinator: Codeunit "PW Batch Coordinator"; + Items: List of [Text]; + Payload: JsonObject; + begin + Items.Add('X'); + Payload.Add('$Items', 'clash'); + + asserterror Coordinator.RunForList("PW Worker Type"::TestWorker, Items, Payload); + + LibraryAssert.ExpectedError('reserved'); + end; + + [Test] + procedure RunForRecordsRejectsReservedPayloadKey() + // [SCENARIO 15.4] RunForRecords rejects a payload containing a $-prefixed key + var + Coordinator: Codeunit "PW Batch Coordinator"; + Batch: Record "PW Batch"; + RecRef: RecordRef; + Payload: JsonObject; + begin + RecRef.GetTable(Batch); + Payload.Add('$TableNo', 99); + + asserterror Coordinator.RunForRecords("PW Worker Type"::TestRecordWorker, RecRef, Payload); + + LibraryAssert.ExpectedError('reserved'); + end; + + #endregion + + local procedure NewLine(): Text + var + LineFeed: Char; + begin + LineFeed := 10; + exit(Format(LineFeed)); + end; } diff --git a/test/src/Tests/PWIntegrationTest.Codeunit.al b/test/src/Tests/PWIntegrationTest.Codeunit.al index 852c2cf..fa7b608 100644 --- a/test/src/Tests/PWIntegrationTest.Codeunit.al +++ b/test/src/Tests/PWIntegrationTest.Codeunit.al @@ -1299,4 +1299,77 @@ codeunit 99209 "PW Integration Test" end; #endregion + + #region Session tracking & mid-batch cleanup + + [Test] + procedure SessionIdStoredOnChunksAtStart() + // [SCENARIO 17.1] RunForList persists each chunk's session id before returning, + // so dead-session recovery also covers chunks that never reach Running + var + Coordinator: Codeunit "PW Batch Coordinator"; + Chunk: Record "PW Batch Chunk"; + Items: List of [Text]; + Payload: JsonObject; + BatchId: Guid; + begin + Items.Add('A'); + Items.Add('B'); + Payload.Add('WorkDurationMs', 1000); + + BatchId := Coordinator + .SetThreads(2) + .SetBatchTimeout(30) + .RunForList("PW Worker Type"::Sample, Items, Payload); + + Chunk.SetRange("Batch Id", BatchId); + Chunk.SetRange("Session Id", 0); + LibraryAssert.IsTrue(Chunk.IsEmpty(), 'Every chunk should carry a session id right after RunForList'); + + Coordinator.WaitForCompletion(BatchId); + Coordinator.Cleanup(BatchId); + Commit(); + end; + + [Test] + procedure CleanupWhileSessionsRunningIsSafe() + // [SCENARIO 17.2] Cleanup during a live batch removes all records; orphaned + // sessions end quietly and the framework keeps working afterwards + var + Coordinator: Codeunit "PW Batch Coordinator"; + Batch: Record "PW Batch"; + Chunk: Record "PW Batch Chunk"; + Items: List of [Text]; + Payload: JsonObject; + Results: List of [JsonObject]; + BatchId: Guid; + begin + Items.Add('Slow'); + Payload.Add('WorkDurationMs', 2000); + BatchId := Coordinator + .SetThreads(1) + .RunForList("PW Worker Type"::Sample, Items, Payload); + + Coordinator.Cleanup(BatchId); + Commit(); + + LibraryAssert.IsFalse(Batch.Get(BatchId), 'Batch record should be gone'); + Chunk.SetRange("Batch Id", BatchId); + LibraryAssert.AreEqual(0, Chunk.Count(), 'Chunk records should be gone'); + + // Give the orphaned session time to hit the missing-record guards and end quietly + Sleep(3000); + + Clear(Items); + Items.Add('X'); + LibraryAssert.IsTrue( + Coordinator.SetThreads(1).SetBatchTimeout(30).RunAndWaitForList( + "PW Worker Type"::TestWorker, Items, Results), + 'A batch started after a mid-flight cleanup should still complete'); + LibraryAssert.AreEqual(1, Results.Count(), 'Should have 1 result'); + + Commit(); + end; + + #endregion }