cli: select transport via TINY_NATS_TRANSPORT (core default, jetstream opt-in)

Defines transport.Transport interface — both NATS (core) and
JetStream (durable) satisfy it. cli/run.go picks at boot based on
TINY_NATS_TRANSPORT env, defaulting to 'core' for backward
compatibility. Setting 'jetstream' ensures the module-edges stream
on startup and binds the JS-backed sender/receiver.

No change for existing deployments — env unset = today's core
req/reply. Flip to durable wire on a per-cluster basis by setting
TINY_NATS_TRANSPORT=jetstream in operator.release_values.

Author: tpoxa

cli/run.go

@@ -352,12 +352,27 @@ var runCmd = &cobra.Command{
 
 		// NATS transport selector. When TINY_NATS_URL is set we route
 		// cross-module messages through NATS subjects instead of the
-		// gRPC AddressPool. Same runner.Handler contract — scheduler
-		// doesn't notice. Unset = legacy gRPC path.
+		// gRPC AddressPool. TINY_NATS_TRANSPORT picks between:
+		//   "core"      — core req/reply, ephemeral, fast (default)
+		//   "jetstream" — durable WorkQueue stream, AckWait pod-death
+		//                 recovery, single-shot on handler error
+		// Unset TINY_NATS_URL = legacy gRPC path, no change.
 		natsRt := connectNATS(ctx, l)
-		var natsTransport *transport.NATS
+		var natsTransport transport.Transport
 		if natsRt != nil {
-			natsTransport = transport.NewNATS(natsRt.NC, moduleInfo.GetNameSanitised(), l)
+			moduleName := moduleInfo.GetNameSanitised()
+			switch os.Getenv("TINY_NATS_TRANSPORT") {
+			case "jetstream":
+				if err := transport.EnsureEdgeStream(ctx, natsRt.JS); err != nil {
+					l.Error(err, "ensure edge stream")
+					os.Exit(1)
+				}
+				natsTransport = transport.NewJetStream(natsRt.JS, natsRt.NC, moduleName, l)
+				l.Info("transport: jetstream-backed durable wire")
+			default:
+				natsTransport = transport.NewNATS(natsRt.NC, moduleName, l)
+				l.Info("transport: core req/reply")
+			}
 		}
 
 		//

internal/transport/nats.go

@@ -58,6 +58,16 @@ func SubjectFor(moduleName string) string {
 	return fmt.Sprintf("%s.%s.msg", subjectPrefix, moduleName)
 }
 
+// Transport is the cross-module wire contract that both the core
+// req/reply NATS transport (this file) and the JetStream-backed
+// transport (jetstream.go) satisfy. cli/run.go picks one at boot
+// based on TINY_NATS_TRANSPORT and binds it to the scheduler — the
+// rest of the SDK doesn't care which substrate carries the bytes.
+type Transport interface {
+	Handler(ctx context.Context, msg *runner.Msg) ([]byte, error)
+	StartReceiver(ctx context.Context, handler runner.Handler) error
+}
+
 // NATS holds the cross-module wire backed by NATS core request/reply.
 // Implements both sender (Handler) and receiver (StartReceiver) so a
 // single instance replaces both AddressPool and the gRPC server.