- TupleId internalized

- Shutdown timeout

Author: et1975

RELEASE_NOTES.md

@@ -1,6 +1,11 @@
+#### 5.1.0 - Apr 2026
+* Breaking: hide TupleId as internal
+* introduce SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS that defaults to TOPOLOGY_MESSAGE_TIMEOUT_SECS
+
 #### 5.0.0 - Mar 2026
 * split Multilang support into a separate package
 * .net 10 and Disruptor 6
+* Timeout setting now affects the tuple expiry
 * self-hosting: executor abstraction and standard .net telemetry
 
 #### 4.1.2 - Sep 2023

docs/content/architecture.fsx

@@ -150,7 +150,7 @@ stateDiagram-v2
 
 1. **System** — stop timers
 2. **Spouts** — stop generating messages (deactivate)
-3. **Sleep(timeout)** — allow in-flight tuples to drain through the bolt DAG
+3. **Sleep(timeout)** — allow in-flight tuples to drain through the bolt DAG (`SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS`, defaults to `TOPOLOGY_MESSAGE_TIMEOUT_SECS`)
 4. **Bolts** — stop processing (deactivate)
 5. **Ackers** — stop tracking (last, so late acks can still be processed)
 

docs/content/guaranteed.fsx

@@ -74,7 +74,7 @@ let source =
                 return! loop <|
                        match cmd, nacked with
                        | Get rc, [] ->
-                            let tupleId,number = Named(string(nextId())), rnd.Next(0, 100)
+                            let tupleId,number = TupleId.OfString(string(nextId())), rnd.Next(0, 100)
                             pending.Add(tupleId,number)
                             rc.Reply(tupleId,number)
                             []

docs/content/self-hosting.fsx

@@ -150,7 +150,8 @@ Configuration
 | `TOPOLOGY_ACKER_TASKS` | 4 | Number of acker task instances (logical units with independent state) |
 | `TOPOLOGY_ACKER_EXECUTORS` | 2 | Number of acker executor threads (tasks are distributed round-robin) |
 | `TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE` | 256 | Base Disruptor ring buffer size (scaled by tasks per executor) |
-| `TOPOLOGY_MESSAGE_TIMEOUT_SECS` | 30 | Timeout for tuple completion; controls acker bucket rotation interval and drain window during shutdown |
+| `TOPOLOGY_MESSAGE_TIMEOUT_SECS` | 30 | Timeout for tuple completion; controls acker bucket rotation interval |
+| `SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS` | = `TOPOLOGY_MESSAGE_TIMEOUT_SECS` | Drain window during shutdown (time between stopping spouts and stopping bolts) |
 | `TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS` | 100 | Spout executor timeout: how often the spout wakes to poll for new tuples when idle |
 | `TOPOLOGY_DEBUG` | false | Enable trace-level logging with timing |
 | `TOPOLOGY_TICK_TUPLE_FREQ_SECS` | (none) | Per-bolt tick tuple interval in seconds |
@@ -198,13 +199,14 @@ The ring buffer must also be large enough to hold `TOPOLOGY_MAX_SPOUT_PENDING` i
 
 **Message timeout and shutdown:**
 
-`TOPOLOGY_MESSAGE_TIMEOUT_SECS` serves three roles:
+`TOPOLOGY_MESSAGE_TIMEOUT_SECS` serves two roles:
 
 1. **Acker bucket rotation** — tuples older than this are expired
-2. **Shutdown drain window** — the system sleeps this long between stopping spouts and stopping bolts
-3. **Restart backoff ceiling** — indirectly affects how long restarts take
+2. **Restart backoff ceiling** — indirectly affects how long restarts take
 
-Lower values speed up shutdown and failure detection, but risk false-expiring tuples that are merely slow to process.
+`SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS` controls the **shutdown drain window** — the sleep between stopping spouts and stopping bolts. It defaults to `TOPOLOGY_MESSAGE_TIMEOUT_SECS` when not set, so existing configurations behave identically. Set it independently when you need a long message timeout (for slow processing) but a fast shutdown.
+
+Lower `TOPOLOGY_MESSAGE_TIMEOUT_SECS` speeds up failure detection but risks false-expiring tuples that are merely slow to process.
 
 **Acker capacity:**
 
@@ -220,6 +222,7 @@ Acker tasks are stateless relative to each other (each tracks a disjoint set of
 | Reduce thread count | Set `withExecutors` lower than `withParallelism` (share threads) |
 | Handle slow consumers | Increase `TOPOLOGY_MAX_SPOUT_PENDING` and `TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE` proportionally |
 | Faster failure detection | Lower `TOPOLOGY_MESSAGE_TIMEOUT_SECS` (but ensure it exceeds your slowest bolt processing time) |
+| Faster shutdown | Set `SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS` independently of message timeout |
 | Reduce idle CPU | Increase `TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS` (default 100ms is usually fine) |
 | Handle high acker load | Increase `TOPOLOGY_ACKER_TASKS` and/or `TOPOLOGY_ACKER_EXECUTORS` |
 

samples/Guaranteed/Topology.fs

@@ -51,7 +51,7 @@ let source =
                 let! cmd = inbox.Receive()
                 return! loop <| match cmd, nacked with
                                 | Get rc, [] ->
-                                    let tupleId,number = Named(string(nextId())), rnd.Next(0, 100)
+                                    let tupleId,number = TupleId.ofString(string(nextId())), rnd.Next(0, 100)
                                     pending.Add(tupleId,number)
                                     rc.Reply(tupleId,number)
                                     []

samples/WordCount/Topology.fs

@@ -18,7 +18,7 @@ let sentences source =
 // so that it could be replayed in case of downstream failure
 let reliableSentences source = 
     let sentence = source()
-    Some(Named sentence, Sentence sentence) // we'll just pretend we've generated a unique Id
+    Some(TupleId.ofString sentence, Sentence sentence) // we'll just pretend we've generated a unique Id
 
 // split bolt - consumes sentences and emits words
 let splitIntoWords (input, emit) = 

src/FsShelter.Multilang/IO/JsonIO.fs

@@ -31,14 +31,14 @@ let private toEmit streamRW t (tid:TupleId option,anchors:TupleId list,stream:st
         if not (List.isEmpty anchors) then
             w.WritePropertyName("anchors")
             w.WriteStartArray()
-            anchors |> List.iter (fun a -> w.WriteValue(string a))
+            anchors |> List.iter (fun a -> w.WriteValue(TupleId.toString a))
             w.WriteEndArray()
 
     let writeId (w:JsonTextWriter) =
         match tid with
         | Some tid -> 
             w.WritePropertyName("id")
-            w.WriteValue(string tid)
+            w.WriteValue(TupleId.toString tid)
         | _ -> ()
 
     let writeTask (w:JsonTextWriter) =
@@ -117,8 +117,8 @@ let private (|Control|_|) (o:JObject) =
     | true, Some p ->
         match p.ToObject() with
         | "next" -> Some (Next)
-        | "ack" -> Some (Ack (Named (o.["id"].ToObject())))
-        | "fail" -> Some (Nack (Named (o.["id"].ToObject())))
+        | "ack" -> Some (Ack (TupleId.ofString (o.["id"].ToObject())))
+        | "fail" -> Some (Nack (TupleId.ofString (o.["id"].ToObject())))
         | "activate" -> Some(Activate)
         | "deactivate" -> Some(Deactivate)
         | _ -> None
@@ -133,7 +133,7 @@ let private (|Stream|_|) findConstructor (o:JObject) =
              let xs = o.["tuple"].Children().GetEnumerator()
              let constr = findConstructor streamId <| fun t -> xs.MoveNext() |> ignore; xs.Current.ToObject(t)
              let comp = if isNull o.["comp"] then "" else o.["comp"].ToObject()
-             Some (InCommand.Tuple(constr(), Named (o.["id"].ToObject()), comp, streamId, o.["task"].ToObject()))
+             Some (InCommand.Tuple(constr(), TupleId.ofString (o.["id"].ToObject()), comp, streamId, o.["task"].ToObject()))
     | _ -> None
 
 let private toCommand (findConstructor:string->FieldReader->unit->'t) str : InCommand<'t> =
@@ -159,8 +159,8 @@ let startWith (stdin:TextReader,stdout:TextWriter) syncOut (log:Log) :Topology.I
         match cmd with
         | Sync -> """{"command":"sync"}"""
         | Pid pid -> sprintf """{"pid":%d}""" pid
-        | Fail tid -> sprintf """{"command":"fail","id":"%s"}""" (string tid)
-        | Ok tid -> sprintf """{"command":"ack","id":"%s"}""" (string tid)
+        | Fail tid -> sprintf """{"command":"fail","id":"%s"}""" (TupleId.toString tid)
+        | Ok tid -> sprintf """{"command":"ack","id":"%s"}""" (TupleId.toString tid)
         | Log (msg,lvl) -> toLog msg (int lvl)
         | Error (msg,ex) -> toLog (sprintf "%s: %s" msg (Exception.toString ex)) (int LogLevel.Error)
         | Emit (t,tid,anchors,stream, task, needTaskIds) -> toEmit streamRW t (tid, anchors, stream, task, needTaskIds)

src/FsShelter.Multilang/IO/ProtoIO.fs

@@ -140,15 +140,15 @@ let private toCommand log (findConstructor:string->FieldReader->unit->'t) (msg:M
     let toConf conf = conf |> Seq.map (fun (x:Collections.Generic.KeyValuePair<string,VL>) -> x.Key, ofValue x.Value) |> Map
     let toContext (ctx:Messages.Context) = { ComponentId = ctx.ComponentId; TaskId = ctx.TaskId; Components = ctx.TaskComponents |> Seq.map (|KeyValue|) |> Map }
     match msg.MsgCase with
-    | Messages.StormMsg.MsgOneofCase.AckCmd -> Ack (Named msg.AckCmd.Id)
+    | Messages.StormMsg.MsgOneofCase.AckCmd -> Ack (TupleId.ofString msg.AckCmd.Id)
     | Messages.StormMsg.MsgOneofCase.Handshake -> Handshake (toConf msg.Handshake.Config, msg.Handshake.PidDir, (toContext msg.Handshake.Context))
     | Messages.StormMsg.MsgOneofCase.Heartbeat -> Heartbeat
-    | Messages.StormMsg.MsgOneofCase.NackCmd -> Nack (Named msg.NackCmd.Id)
+    | Messages.StormMsg.MsgOneofCase.NackCmd -> Nack (TupleId.ofString msg.NackCmd.Id)
     | Messages.StormMsg.MsgOneofCase.NextCmd -> Next
     | Messages.StormMsg.MsgOneofCase.TaskIds -> TaskIds (msg.TaskIds.TaskIds |> List.ofSeq)
     | Messages.StormMsg.MsgOneofCase.StreamIn ->
         let constr = findConstructor msg.StreamIn.Stream
-        InCommand.Tuple ((ofFields constr msg.StreamIn.Tuple)(), Named msg.StreamIn.Id, msg.StreamIn.Comp, msg.StreamIn.Stream, msg.StreamIn.Task)
+        InCommand.Tuple ((ofFields constr msg.StreamIn.Tuple)(), TupleId.ofString msg.StreamIn.Id, msg.StreamIn.Comp, msg.StreamIn.Stream, msg.StreamIn.Task)
     | Messages.StormMsg.MsgOneofCase.ActivateCmd -> Activate
     | Messages.StormMsg.MsgOneofCase.DeactivateCmd -> Deactivate
     | _ -> failwithf "Unexpected command: %A" msg
@@ -165,17 +165,17 @@ let startWith (stdin:#Stream,stdout:#Stream) syncOut (log:Log) :Topology.IO<'t>
         match cmd with
         | Sync -> Messages.ShellMsg(Sync = Messages.SyncReply())
         | Pid pid -> Messages.ShellMsg(Pid = Messages.PidReply(Pid = pid))
-        | Fail tid -> Messages.ShellMsg(Fail = Messages.FailReply(Id = string tid))
-        | Ok tid -> Messages.ShellMsg(Ok = Messages.OkReply(Id = string tid))
+        | Fail tid -> Messages.ShellMsg(Fail = Messages.FailReply(Id = TupleId.toString tid))
+        | Ok tid -> Messages.ShellMsg(Ok = Messages.OkReply(Id = TupleId.toString tid))
         | Log (msg,lvl) -> Messages.ShellMsg(Log = Messages.LogCommand(Text = msg, Level = enum (int lvl)))
         | Error (msg,ex) -> Messages.ShellMsg(Log = Messages.LogCommand(Text = (sprintf "%s: %s" msg (Exception.toString ex)), Level = Messages.LogCommand.Types.LogLevel.Error))
         | Emit (t,tid,anchors,stream,task,needTaskIds) -> 
             let (_,d) = streamRW |> Map.find stream
             let cmd = Messages.EmitCommand(Stream = stream)
             cmd.Tuple.Add (toFields d t)
-            if Option.isSome tid then cmd.Id <- string tid.Value
+            if Option.isSome tid then cmd.Id <- TupleId.toString tid.Value
             if Option.isSome needTaskIds && needTaskIds.Value then cmd.NeedTaskIds <- needTaskIds.Value
-            if not (List.isEmpty anchors) then cmd.Anchors.Add (anchors |> List.map string)
+            if not (List.isEmpty anchors) then cmd.Anchors.Add (anchors |> List.map TupleId.toString)
             Messages.ShellMsg(Emit = cmd)
         |> write
 

src/FsShelter.Tests/AckerTests.fs

@@ -30,7 +30,7 @@ type Tracker =
 let mkTrackedTopology (tracker: Tracker) maxPending timeout =
     let numbers (t: Tracker) =
         Interlocked.Increment &t.emitted.contents |> ignore
-        Some(Named(string t.emitted.Value), Original { x = 1 })
+        Some(TupleId.ofString(string t.emitted.Value), Original { x = 1 })
 
     let t = topology "acker-test" {
         let s1 = numbers
@@ -107,7 +107,7 @@ let ``Acker nacks on bolt failure`` () =
 
     let failNumbers (t: Tracker) =
         Interlocked.Increment &t.emitted.contents |> ignore
-        Some(Named(string t.emitted.Value), Original { x = 1 })
+        Some(TupleId.ofString(string t.emitted.Value), Original { x = 1 })
 
     let failingBolt (input: Schema, emit: Schema -> unit) =
         Interlocked.Increment failCount |> ignore
@@ -157,7 +157,7 @@ let ``Acker expires timed-out tuples`` () =
         if not emittedOnce.Value then
             emittedOnce := true
             Interlocked.Increment &t.emitted.contents |> ignore
-            Some(Named(string t.emitted.Value), Original { x = 42 })
+            Some(TupleId.ofString(string t.emitted.Value), Original { x = 42 })
         else
             None
 
@@ -210,7 +210,7 @@ let ``Acker capacity overflow nacks tuples`` () =
 
     let fastNumbers (t: Tracker) =
         Interlocked.Increment &t.emitted.contents |> ignore
-        Some(Named(string t.emitted.Value), Original { x = 1 })
+        Some(TupleId.ofString(string t.emitted.Value), Original { x = 1 })
 
     // bot that never acks - will cause acker inFlight to grow
     let blackHole (input: Schema, _: Schema -> unit) = ()

src/FsShelter.Tests/BoltAndGroupingTests.fs

@@ -24,7 +24,7 @@ let ``Shuffle grouping distributes across instances`` () =
     let acked = ref 0L
 
     let numbers (_: unit) =
-        Some(Named(string (Guid.NewGuid())), Original { x = 1 })
+        Some(TupleId.ofString(string (Guid.NewGuid())), Original { x = 1 })
 
     let countingBolt (input: Schema, _: Schema -> unit) =
         let tid = Thread.CurrentThread.ManagedThreadId
@@ -72,7 +72,7 @@ let ``Fields grouping routes same key to same instance`` () =
     let numbersWithKeys (_: unit) =
         let x = int (Interlocked.Increment emitted)
         let key = x % 5  // 5 distinct keys
-        Some(Named(string x), Original { x = key })
+        Some(TupleId.ofString(string x), Original { x = key })
 
     let split' (input, emit) =
         match input with
@@ -135,7 +135,7 @@ let ``All grouping broadcasts to all instances`` () =
 
     let oneNumber (_: unit) =
         if not (Interlocked.Exchange(emittedOnce, true)) then
-            Some(Named "1", Original { x = 42 })
+            Some(TupleId.ofString "1", Original { x = 42 })
         else
             None
 
@@ -194,10 +194,10 @@ let ``Auto-ack bolt sends Ok on success`` () =
                     Map.empty
                     out
 
-    dispatch (InCommand.Tuple(Original {x = 1}, Named "tuple-1", "s1", "Original", 0))
+    dispatch (InCommand.Tuple(Original {x = 1}, TupleId.ofString "tuple-1", "s1", "Original", 0))
 
     let oks = outMsgs |> Seq.choose (function OutCommand.Ok id -> Some id | _ -> None) |> Seq.toList
-    test <@ oks = [Named "tuple-1"] @>
+    test <@ oks = [TupleId.ofString "tuple-1"] @>
 
 // ---------------------------------------------------------------------------
 // Storm BoltTest: testBoltFailOnException
@@ -221,10 +221,10 @@ let ``Auto-ack bolt sends Fail on exception`` () =
                     Map.empty
                     out
 
-    dispatch (InCommand.Tuple(Original {x = 1}, Named "tuple-1", "s1", "Original", 0))
+    dispatch (InCommand.Tuple(Original {x = 1}, TupleId.ofString "tuple-1", "s1", "Original", 0))
 
     let fails = outMsgs |> Seq.choose (function OutCommand.Fail id -> Some id | _ -> None) |> Seq.toList
-    test <@ fails = [Named "tuple-1"] @>
+    test <@ fails = [TupleId.ofString "tuple-1"] @>
 
 // ---------------------------------------------------------------------------
 // Storm BoltTest: testBoltAnchoredEmit
@@ -237,7 +237,7 @@ let ``Anchored emit preserves tuple lineage`` () =
 
     let numbers (t: AckerTests.Tracker) =
         Interlocked.Increment &t.emitted.contents |> ignore
-        Some(Named(string t.emitted.Value), Original { x = 1 })
+        Some(TupleId.ofString(string t.emitted.Value), Original { x = 1 })
 
     let anchoredPassthrough (input: Schema, emit: Schema -> unit) =
         match input with