Verified feat: init base draft support window

1 files changed, 66 insertions, 0 deletions

diff --git a/runtime/task.go b/runtime/task.go
new file mode 100644
index 0000000..734976e
--- /dev/null
+++ b/runtime/task.go
@@ -0,0 +1,66 @@
+package runtime
+
+import (
+        "context"
+        "fmt"
+        "time"
+
+        "flux/core/operator"
+        "flux/transport"
+)
+
+// RunTask runs a stream task in the background. The returned channel yields the
+// first error that stopped it (context.Canceled when ctx ends early) and is
+// closed when the task is done, so callers can join on it.
+func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) <-chan error {
+        done := make(chan error, 1)
+
+        go func() {
+                defer close(done)
+                if err := runTask(ctx, input, output, op); err != nil {
+                        done <- err
+                }
+        }()
+
+        return done
+}
+
+func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) (err error) {
+        // ponytail: assumes one writer per output channel. Fan-in needs a refcount
+        // or a dedicated closer, otherwise the second Close panics.
+        defer func() {
+                if cerr := output.Close(); cerr != nil && err == nil {
+                        err = fmt.Errorf("close output: %w", cerr)
+                }
+        }()
+
+        if err := op.Open(ctx); err != nil {
+                return fmt.Errorf("open operator: %w", err)
+        }
+        defer func() {
+                // ctx is usually already cancelled by the time we get here — that is
+                // why the task stopped. Close still needs a live window to flush, so
+                // give it one that keeps ctx's values but drops the cancellation.
+                closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+                defer cancel()
+
+                if cerr := op.Close(closeCtx); cerr != nil && err == nil {
+                        err = fmt.Errorf("close operator: %w", cerr)
+                }
+        }()
+
+        for {
+                record, ok := input.Receive(ctx)
+                if !ok {
+                        break
+                }
+
+                if err := op.Process(ctx, record, output); err != nil {
+                        return fmt.Errorf("process record: %w", err)
+                }
+        }
+
+        // Receive also returns !ok on cancellation, so report that as an error
+        // rather than a clean drain.
+        return ctx.Err()
+}