feat: implement chunked file transfer protocol

This change replaces the simple full-file transfer mechanism with a robust
chunked protocol that supports:
- Concurrent uploads/downloads with unique transfer IDs
- 1MB chunk streaming to support large files
- Transfer state management with acknowledgments
- Round-robin scheduling for fair resource usage during concurrent transfers

Signed-off-by: llogen <christoph.lange@blindspot.software>

Author: llogen

cmds/dutagent/states.go

@@ -29,6 +29,7 @@ type runCmdArgs struct {
 	cmdMsg      *pb.Command
 	dev         dut.Device
 	cmd         dut.Command
+	broker      *dutagent.Broker
 	session     module.Session
 	moduleErrCh chan error
 	brokerErrCh <-chan error
@@ -98,8 +99,10 @@ func findDUTCmd(_ context.Context, args runCmdArgs) (runCmdArgs, fsm.State[runCm
 // in a separate goroutine, this state will not wait for the execution to finish.
 // Further, worker goroutines will be started to serve the module-to-client communication
 // during the module execution.
+//
+//nolint:funlen
 func executeModules(ctx context.Context, args runCmdArgs) (runCmdArgs, fsm.State[runCmdArgs], error) {
-	broker := &dutagent.Broker{}
+	args.broker = &dutagent.Broker{}
 
 	// Deferred initialization of the moduleErr channel: only create if not already provided
 	// (tests may still pass a custom channel).
@@ -110,19 +113,26 @@ func executeModules(ctx context.Context, args runCmdArgs) (runCmdArgs, fsm.State
 	rpcCtx := ctx
 	modCtx, modCtxCancel := context.WithCancel(rpcCtx)
 
-	moduleSession, brokerErrCh := broker.Start(modCtx, args.stream)
+	moduleSession, brokerErrCh := args.broker.Start(modCtx, args.stream)
 	args.brokerErrCh = brokerErrCh
 	args.session = moduleSession
 
 	// Run the modules in a goroutine.
 	// Termination of the module execution is signaled by closing the moduleErrCh channel.
 	go func() {
+		defer modCtxCancel() // Ensure workers exit even if stream doesn't close
+
 		cnt := len(args.cmd.Modules)
 
 		for idx, module := range args.cmd.Modules {
 			if ctx.Err() != nil {
 				log.Printf("Execution aborted, %d of %d modules done: %v", idx, cnt, ctx.Err())
-				modCtxCancel()
+				args.broker.Shutdown()
+
+				// Wait for file transfers to complete (workers will exit gracefully)
+				log.Print("Waiting for file transfers to complete...")
+				<-args.broker.WaitForTransfersToComplete()
+				log.Print("All file transfers completed")
 
 				return
 			}
@@ -141,14 +151,25 @@ func executeModules(ctx context.Context, args runCmdArgs) (runCmdArgs, fsm.State
 				args.moduleErrCh <- err
 
 				log.Printf("Module %q failed: %v", module.Config.Name, err)
-				modCtxCancel()
+				args.broker.Shutdown()
+
+				// Wait for file transfers to complete (workers will exit gracefully)
+				log.Print("Waiting for file transfers to complete...")
+				<-args.broker.WaitForTransfersToComplete()
+				log.Print("All file transfers completed")
 
 				return
 			}
 		}
 
 		log.Print("All modules finished successfully")
-		modCtxCancel()
+		args.broker.Shutdown()
+
+		// Wait for file transfers to complete (workers will exit gracefully)
+		log.Print("Waiting for file transfers to complete...")
+		<-args.broker.WaitForTransfersToComplete()
+		log.Print("All file transfers completed")
+
 		close(args.moduleErrCh)
 	}()