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/dutctl/file_transfer.go

@@ -10,9 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"io/fs"
 	"log"
-	"os"
 	"strings"
 
 	"connectrpc.com/connect"
@@ -101,6 +99,8 @@ func (app *application) runRPC(device, command string, cmdArgs []string) error {
 
 	errChan := make(chan error, numWorkers)
 
+	ftManager := newClientFileTransferManager()
+
 	stream := app.rpcClient.Run(runCtx)
 	req := &pb.RunRequest{
 		Msg: &pb.RunRequest_Command{
@@ -175,51 +175,30 @@ func (app *application) runRPC(device, command string, cmdArgs []string) error {
 				case *pb.Console_Stdin:
 					log.Printf("Unexpected Console Stdin %q", string(consoleData.Stdin))
 				}
-			case *pb.RunResponse_FileRequest:
-				path := msg.FileRequest.GetPath()
-				log.Printf("File request for: %q\n", path)
+			case *pb.RunResponse_FileTransferRequest:
+				ftReq := msg.FileTransferRequest
 
-				content, err := os.ReadFile(path)
-				if err != nil {
-					errChan <- fmt.Errorf("reading requested file %q: %w", path, err)
+				ftErr := ftManager.handleFileTransferRequest(ftReq, stream)
+				if ftErr != nil {
+					errChan <- ftErr
 
 					return
 				}
 
-				err = stream.Send(&pb.RunRequest{
-					Msg: &pb.RunRequest_File{
-						File: &pb.File{
-							Path:    path,
-							Content: content,
-						},
-					},
-				})
-				if err != nil {
-					errChan <- fmt.Errorf("sending requested file %q: %w", path, err)
-
-					return
-				}
+			case *pb.RunResponse_FileChunk:
+				chunk := msg.FileChunk
 
-				log.Printf("Sent file: %q\n", path)
-			case *pb.RunResponse_File:
-				path := msg.File.GetPath()
-				content := msg.File.GetContent()
-
-				log.Printf("Received file: %q\n", path)
-
-				if len(content) == 0 {
-					log.Println("Received empty file content")
-				}
-
-				perm := 0600
-
-				err = os.WriteFile(path, content, fs.FileMode(perm))
-				if err != nil {
-					errChan <- fmt.Errorf("saving received file %q: %w", path, err)
+				chunkErr := ftManager.handleFileChunk(chunk, stream)
+				if chunkErr != nil {
+					errChan <- chunkErr
 
 					return
 				}
 
+			case *pb.RunResponse_FileTransferResponse:
+				ftRes := msg.FileTransferResponse
+				ftManager.handleFileTransferResponse(ftRes)
+
 			default:
 				log.Printf("Unexpected message type %T", msg)
 			}

cmds/dutctl/rpc.go

@@ -7,6 +7,7 @@ require (
 	github.com/bougou/go-ipmi v0.7.8
 	github.com/go-playground/validator/v10 v10.30.1
 	github.com/google/go-cmp v0.7.0
+	github.com/google/uuid v1.1.2
 	github.com/stianeikeland/go-rpio/v4 v4.6.0
 	github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07
 	golang.org/x/crypto v0.47.0
@@ -20,7 +21,6 @@ require (
 	github.com/gabriel-vasile/mimetype v1.4.12 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
-	github.com/google/uuid v1.1.2 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect

go.mod

@@ -30,12 +30,14 @@ type Broker struct {
 func (b *Broker) init() {
 	log.Print("Broker: Initializing")
 
-	b.session.printCh = make(chan string)
-	b.session.stdinCh = make(chan []byte)
-	b.session.stdoutCh = make(chan []byte)
-	b.session.stderrCh = make(chan []byte)
-	b.session.fileReqCh = make(chan string)
-	b.session.fileCh = make(chan chan []byte)
+	b.session.printCh = make(chan string, channelBufferSize)
+	b.session.stdinCh = make(chan []byte, channelBufferSize)
+	b.session.stdoutCh = make(chan []byte, channelBufferSize)
+	b.session.stderrCh = make(chan []byte, channelBufferSize)
+	b.session.activeUploads = make(map[string]*uploadState)
+	b.session.activeDownloads = make(map[string]*downloadState)
+	b.session.downloadOrder = make([]string, 0)
+	b.session.downloadIndex = 0
 
 	// Buffer equals number of workers so error sends never block.
 	b.errCh = make(chan error, numWorkers)