feat: create working client

1 files changed, 1024 insertions, 0 deletions

diff --git a/client_test.go b/client_test.go
new file mode 100644
index 0000000..af890f0
--- /dev/null
+++ b/client_test.go
@@ -0,0 +1,1024 @@
+package zw
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"net"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	zwdaemonv1 "git.pablu.de/pablu/zw-go/gen/zwdaemon/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+const testZwErrorDetailTypeURL = "type.googleapis.com/zw.daemon.v1.ZwErrorDetail"
+
+type testService struct {
+	zwdaemonv1.UnimplementedZwDaemonServiceServer
+
+	response *zwdaemonv1.GetServerInfoResponse
+	err      error
+
+	createRequests chan *zwdaemonv1.CreateRequest
+	createErr      error
+	unsealRequests chan *zwdaemonv1.UnsealRequest
+	unsealErr      error
+	sealCalls      chan struct{}
+	sealErr        error
+
+	listFilesRequests chan *zwdaemonv1.ListFilesRequest
+	listFilesResponse *zwdaemonv1.ListFilesResponse
+	listFilesErr      error
+
+	deleteFileRequests chan *zwdaemonv1.DeleteFileRequest
+	deleteFileErr      error
+	compactCalls       chan struct{}
+	compactErr         error
+
+	getFile      func(*zwdaemonv1.GetFileRequest, grpc.ServerStreamingServer[zwdaemonv1.GetFileResponse]) error
+	getFileRange func(*zwdaemonv1.GetFileRangeRequest, grpc.ServerStreamingServer[zwdaemonv1.GetFileRangeResponse]) error
+	addFile      func(grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse]) error
+}
+
+func (s *testService) GetServerInfo(
+	context.Context,
+	*zwdaemonv1.GetServerInfoRequest,
+) (*zwdaemonv1.GetServerInfoResponse, error) {
+	return s.response, s.err
+}
+
+func (s *testService) Create(
+	_ context.Context,
+	request *zwdaemonv1.CreateRequest,
+) (*zwdaemonv1.CreateResponse, error) {
+	if s.createRequests != nil {
+		s.createRequests <- request
+	}
+	return &zwdaemonv1.CreateResponse{}, s.createErr
+}
+
+func (s *testService) Unseal(
+	_ context.Context,
+	request *zwdaemonv1.UnsealRequest,
+) (*zwdaemonv1.UnsealResponse, error) {
+	if s.unsealRequests != nil {
+		s.unsealRequests <- request
+	}
+	return &zwdaemonv1.UnsealResponse{}, s.unsealErr
+}
+
+func (s *testService) Seal(
+	context.Context,
+	*zwdaemonv1.SealRequest,
+) (*zwdaemonv1.SealResponse, error) {
+	if s.sealCalls != nil {
+		s.sealCalls <- struct{}{}
+	}
+	return &zwdaemonv1.SealResponse{}, s.sealErr
+}
+
+func (s *testService) ListFiles(
+	_ context.Context,
+	request *zwdaemonv1.ListFilesRequest,
+) (*zwdaemonv1.ListFilesResponse, error) {
+	if s.listFilesRequests != nil {
+		s.listFilesRequests <- request
+	}
+	if s.listFilesResponse == nil {
+		return &zwdaemonv1.ListFilesResponse{}, s.listFilesErr
+	}
+	return s.listFilesResponse, s.listFilesErr
+}
+
+func (s *testService) DeleteFile(
+	_ context.Context,
+	request *zwdaemonv1.DeleteFileRequest,
+) (*zwdaemonv1.DeleteFileResponse, error) {
+	if s.deleteFileRequests != nil {
+		s.deleteFileRequests <- request
+	}
+	return &zwdaemonv1.DeleteFileResponse{}, s.deleteFileErr
+}
+
+func (s *testService) Compact(
+	context.Context,
+	*zwdaemonv1.CompactRequest,
+) (*zwdaemonv1.CompactResponse, error) {
+	if s.compactCalls != nil {
+		s.compactCalls <- struct{}{}
+	}
+	return &zwdaemonv1.CompactResponse{}, s.compactErr
+}
+
+func (s *testService) GetFile(
+	request *zwdaemonv1.GetFileRequest,
+	stream grpc.ServerStreamingServer[zwdaemonv1.GetFileResponse],
+) error {
+	if s.getFile == nil {
+		return nil
+	}
+	return s.getFile(request, stream)
+}
+
+func (s *testService) GetFileRange(
+	request *zwdaemonv1.GetFileRangeRequest,
+	stream grpc.ServerStreamingServer[zwdaemonv1.GetFileRangeResponse],
+) error {
+	if s.getFileRange == nil {
+		return nil
+	}
+	return s.getFileRange(request, stream)
+}
+
+func (s *testService) AddFile(
+	stream grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse],
+) error {
+	if s.addFile == nil {
+		return stream.SendAndClose(&zwdaemonv1.AddFileResponse{})
+	}
+	return s.addFile(stream)
+}
+
+func startTestServer(t *testing.T, service *testService) string {
+	t.Helper()
+
+	socketPath := filepath.Join(t.TempDir(), "daemon.sock")
+	listener, err := net.Listen("unix", socketPath)
+	if err != nil {
+		t.Fatalf("listen on Unix socket: %v", err)
+	}
+
+	server := grpc.NewServer()
+	zwdaemonv1.RegisterZwDaemonServiceServer(server, service)
+
+	serveErr := make(chan error, 1)
+	go func() {
+		serveErr <- server.Serve(listener)
+	}()
+
+	t.Cleanup(func() {
+		server.Stop()
+		if err := <-serveErr; err != nil {
+			t.Errorf("serve gRPC test server: %v", err)
+		}
+	})
+
+	return socketPath
+}
+
+func connectTestClient(t *testing.T, socketPath string) (*Client, error) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	return Connect(ctx, socketPath)
+}
+
+func startConnectedTestClient(t *testing.T, service *testService) *Client {
+	t.Helper()
+
+	service.response = &zwdaemonv1.GetServerInfoResponse{ApiMajor: supportedAPIMajor}
+	client, err := connectTestClient(t, startTestServer(t, service))
+	if err != nil {
+		t.Fatalf("Connect() error = %v", err)
+	}
+	t.Cleanup(func() { _ = client.Close() })
+	return client
+}
+
+func TestConnectAcceptsSupportedMajorAndMinorDifferences(t *testing.T) {
+	for _, minor := range []uint32{0, 99} {
+		t.Run(fmt.Sprintf("minor_%d", minor), func(t *testing.T) {
+			socketPath := startTestServer(t, &testService{
+				response: &zwdaemonv1.GetServerInfoResponse{
+					ApiMajor:      supportedAPIMajor,
+					ApiMinor:      minor,
+					DaemonVersion: "test-version",
+				},
+			})
+
+			client, err := connectTestClient(t, socketPath)
+			if err != nil {
+				t.Fatalf("Connect() error = %v", err)
+			}
+			t.Cleanup(func() { _ = client.Close() })
+		})
+	}
+}
+
+func TestConnectRejectsUnsupportedMajor(t *testing.T) {
+	socketPath := startTestServer(t, &testService{
+		response: &zwdaemonv1.GetServerInfoResponse{
+			ApiMajor: 2,
+			ApiMinor: 3,
+		},
+	})
+
+	client, err := connectTestClient(t, socketPath)
+	if client != nil {
+		_ = client.Close()
+		t.Fatal("Connect() returned a client for an unsupported API major")
+	}
+
+	var versionErr *IncompatibleAPIVersionError
+	if !errors.As(err, &versionErr) {
+		t.Fatalf("Connect() error = %T %v, want *IncompatibleAPIVersionError", err, err)
+	}
+	if versionErr.ServerMajor != 2 || versionErr.ServerMinor != 3 || versionErr.SupportedMajor != 1 {
+		t.Fatalf("version error = %#v", versionErr)
+	}
+}
+
+func TestServerInfoMapsResponse(t *testing.T) {
+	socketPath := startTestServer(t, &testService{
+		response: &zwdaemonv1.GetServerInfoResponse{
+			ApiMajor:      1,
+			ApiMinor:      7,
+			DaemonVersion: "1.2.3-test",
+		},
+	})
+
+	client, err := connectTestClient(t, socketPath)
+	if err != nil {
+		t.Fatalf("Connect() error = %v", err)
+	}
+	t.Cleanup(func() { _ = client.Close() })
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	info, err := client.ServerInfo(ctx)
+	if err != nil {
+		t.Fatalf("ServerInfo() error = %v", err)
+	}
+
+	want := ServerInfo{APIMajor: 1, APIMinor: 7, DaemonVersion: "1.2.3-test"}
+	if info != want {
+		t.Fatalf("ServerInfo() = %#v, want %#v", info, want)
+	}
+}
+
+func TestConnectPreservesStructuredRPCError(t *testing.T) {
+	serverStatus, err := status.New(codes.FailedPrecondition, "vault is sealed").WithDetails(
+		&zwdaemonv1.ZwErrorDetail{
+			Code:      zwdaemonv1.ZwErrorCode_ZW_ERROR_CODE_VAULT_SEALED,
+			Retryable: true,
+		},
+	)
+	if err != nil {
+		t.Fatalf("create structured status: %v", err)
+	}
+	socketPath := startTestServer(t, &testService{err: serverStatus.Err()})
+
+	_, err = connectTestClient(t, socketPath)
+	rpcErr := requireRPCError(t, err)
+
+	if rpcErr.GRPCCode != codes.FailedPrecondition {
+		t.Errorf("GRPCCode = %v, want %v", rpcErr.GRPCCode, codes.FailedPrecondition)
+	}
+	if rpcErr.ZwCode == nil || *rpcErr.ZwCode != ErrorCodeVaultSealed {
+		t.Errorf("ZwCode = %v, want %v", rpcErr.ZwCode, ErrorCodeVaultSealed)
+	}
+	if rpcErr.Message != "vault is sealed" {
+		t.Errorf("Message = %q, want %q", rpcErr.Message, "vault is sealed")
+	}
+	if !rpcErr.Retryable {
+		t.Error("Retryable = false, want true")
+	}
+	if got := status.Code(err); got != codes.FailedPrecondition {
+		t.Errorf("status.Code(error) = %v, want %v", got, codes.FailedPrecondition)
+	}
+}
+
+func TestConnectPreservesUnknownZwErrorCode(t *testing.T) {
+	const unknownCode zwdaemonv1.ZwErrorCode = 999
+	serverStatus, err := status.New(codes.Internal, "future daemon error").WithDetails(
+		&zwdaemonv1.ZwErrorDetail{Code: unknownCode},
+	)
+	if err != nil {
+		t.Fatalf("create structured status: %v", err)
+	}
+	socketPath := startTestServer(t, &testService{err: serverStatus.Err()})
+
+	_, err = connectTestClient(t, socketPath)
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.ZwCode == nil || *rpcErr.ZwCode != unknownCode {
+		t.Fatalf("ZwCode = %v, want numeric value %d", rpcErr.ZwCode, unknownCode)
+	}
+}
+
+func TestConnectHandlesStatusWithoutZwDetail(t *testing.T) {
+	socketPath := startTestServer(t, &testService{
+		err: status.Error(codes.Unavailable, "daemon unavailable"),
+	})
+
+	_, err := connectTestClient(t, socketPath)
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.ZwCode != nil {
+		t.Errorf("ZwCode = %v, want nil", rpcErr.ZwCode)
+	}
+	if rpcErr.Retryable {
+		t.Error("Retryable = true without ZwErrorDetail")
+	}
+}
+
+func TestConnectIgnoresMalformedZwErrorDetail(t *testing.T) {
+	serverErr := statusWithRawDetail(
+		codes.Internal,
+		"malformed detail",
+		testZwErrorDetailTypeURL,
+		[]byte{0xff},
+	)
+	socketPath := startTestServer(t, &testService{err: serverErr})
+
+	_, err := connectTestClient(t, socketPath)
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.ZwCode != nil {
+		t.Errorf("ZwCode = %v, want nil", rpcErr.ZwCode)
+	}
+	if rpcErr.GRPCCode != codes.Internal || rpcErr.Message != "malformed detail" {
+		t.Errorf("base status was not preserved: %#v", rpcErr)
+	}
+}
+
+func TestConnectRequiresExactZwErrorDetailTypeURL(t *testing.T) {
+	encoded, err := proto.Marshal(&zwdaemonv1.ZwErrorDetail{
+		Code:      zwdaemonv1.ZwErrorCode_ZW_ERROR_CODE_VAULT_SEALED,
+		Retryable: true,
+	})
+	if err != nil {
+		t.Fatalf("marshal ZwErrorDetail: %v", err)
+	}
+	serverErr := statusWithRawDetail(
+		codes.FailedPrecondition,
+		"wrong type URL prefix",
+		"https://example.invalid/zw.daemon.v1.ZwErrorDetail",
+		encoded,
+	)
+	socketPath := startTestServer(t, &testService{err: serverErr})
+
+	_, err = connectTestClient(t, socketPath)
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.ZwCode != nil {
+		t.Errorf("ZwCode = %v for non-canonical type URL, want nil", rpcErr.ZwCode)
+	}
+	if rpcErr.Retryable {
+		t.Error("Retryable = true for non-canonical type URL")
+	}
+}
+
+func TestCreateAndUnsealMapRequests(t *testing.T) {
+	service := &testService{
+		createRequests: make(chan *zwdaemonv1.CreateRequest, 1),
+		unsealRequests: make(chan *zwdaemonv1.UnsealRequest, 1),
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := client.Create(ctx, "/vaults/new.zw", "create-password"); err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+	createRequest := <-service.createRequests
+	if createRequest.GetVaultPath() != "/vaults/new.zw" {
+		t.Errorf("Create vault path = %q", createRequest.GetVaultPath())
+	}
+	if createRequest.GetPassword() != "create-password" {
+		t.Errorf("Create password = %q", createRequest.GetPassword())
+	}
+
+	if err := client.Unseal(ctx, "/vaults/existing.zw", "unseal-password"); err != nil {
+		t.Fatalf("Unseal() error = %v", err)
+	}
+	unsealRequest := <-service.unsealRequests
+	if unsealRequest.GetVaultPath() != "/vaults/existing.zw" {
+		t.Errorf("Unseal vault path = %q", unsealRequest.GetVaultPath())
+	}
+	if unsealRequest.GetPassword() != "unseal-password" {
+		t.Errorf("Unseal password = %q", unsealRequest.GetPassword())
+	}
+}
+
+func TestSealAndCompactCallDaemon(t *testing.T) {
+	service := &testService{
+		sealCalls:    make(chan struct{}, 1),
+		compactCalls: make(chan struct{}, 1),
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := client.Seal(ctx); err != nil {
+		t.Fatalf("Seal() error = %v", err)
+	}
+	<-service.sealCalls
+
+	if err := client.Compact(ctx); err != nil {
+		t.Fatalf("Compact() error = %v", err)
+	}
+	<-service.compactCalls
+}
+
+func TestListFilesMapsFilterAndPreservesResponseOrder(t *testing.T) {
+	service := &testService{
+		listFilesRequests: make(chan *zwdaemonv1.ListFilesRequest, 1),
+		listFilesResponse: &zwdaemonv1.ListFilesResponse{
+			Files: []*zwdaemonv1.ListedFile{
+				{LogicalPath: "logs/second.txt", Size: 22},
+				{LogicalPath: "logs/first.txt", Size: 11},
+			},
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	files, err := client.ListFiles(ctx, ListFilter{Prefix: "logs/", Contains: "txt"})
+	if err != nil {
+		t.Fatalf("ListFiles() error = %v", err)
+	}
+
+	request := <-service.listFilesRequests
+	if request.Prefix == nil || request.GetPrefix() != "logs/" {
+		t.Errorf("ListFiles prefix = %v", request.Prefix)
+	}
+	if request.Contains == nil || request.GetContains() != "txt" {
+		t.Errorf("ListFiles contains = %v", request.Contains)
+	}
+
+	want := []ListedFile{
+		{LogicalPath: "logs/second.txt", Size: 22},
+		{LogicalPath: "logs/first.txt", Size: 11},
+	}
+	if len(files) != len(want) {
+		t.Fatalf("ListFiles() returned %d files, want %d", len(files), len(want))
+	}
+	for index := range want {
+		if files[index] != want[index] {
+			t.Errorf("ListFiles()[%d] = %#v, want %#v", index, files[index], want[index])
+		}
+	}
+}
+
+func TestListPathsOmitsEmptyFiltersAndMapsPaths(t *testing.T) {
+	service := &testService{
+		listFilesRequests: make(chan *zwdaemonv1.ListFilesRequest, 1),
+		listFilesResponse: &zwdaemonv1.ListFilesResponse{
+			Files: []*zwdaemonv1.ListedFile{
+				{LogicalPath: "first", Size: 10},
+				{LogicalPath: "second", Size: 20},
+			},
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	paths, err := client.ListPaths(ctx)
+	if err != nil {
+		t.Fatalf("ListPaths() error = %v", err)
+	}
+
+	request := <-service.listFilesRequests
+	if request.Prefix != nil || request.Contains != nil {
+		t.Errorf("ListPaths request filters = prefix %v, contains %v; want both absent", request.Prefix, request.Contains)
+	}
+	want := []string{"first", "second"}
+	if len(paths) != len(want) {
+		t.Fatalf("ListPaths() returned %d paths, want %d", len(paths), len(want))
+	}
+	for index := range want {
+		if paths[index] != want[index] {
+			t.Errorf("ListPaths()[%d] = %q, want %q", index, paths[index], want[index])
+		}
+	}
+}
+
+func TestDeleteFileMapsRequestAndStructuredError(t *testing.T) {
+	serverStatus, err := status.New(codes.NotFound, "entry not found").WithDetails(
+		&zwdaemonv1.ZwErrorDetail{
+			Code: zwdaemonv1.ZwErrorCode_ZW_ERROR_CODE_ENTRY_NOT_FOUND,
+		},
+	)
+	if err != nil {
+		t.Fatalf("create structured status: %v", err)
+	}
+	service := &testService{
+		deleteFileRequests: make(chan *zwdaemonv1.DeleteFileRequest, 1),
+		deleteFileErr:      serverStatus.Err(),
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err = client.DeleteFile(ctx, "missing.txt")
+	request := <-service.deleteFileRequests
+	if request.GetLogicalPath() != "missing.txt" {
+		t.Errorf("DeleteFile logical path = %q", request.GetLogicalPath())
+	}
+
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.GRPCCode != codes.NotFound {
+		t.Errorf("GRPCCode = %v, want %v", rpcErr.GRPCCode, codes.NotFound)
+	}
+	if rpcErr.ZwCode == nil || *rpcErr.ZwCode != ErrorCodeEntryNotFound {
+		t.Errorf("ZwCode = %v, want %v", rpcErr.ZwCode, ErrorCodeEntryNotFound)
+	}
+}
+
+func TestConvertRPCErrorAcceptsNil(t *testing.T) {
+	if err := convertRPCError(nil); err != nil {
+		t.Fatalf("convertRPCError(nil) = %v", err)
+	}
+}
+
+func TestGetFileReadsChunksAcrossBuffersAndSkipsEmptyChunks(t *testing.T) {
+	requests := make(chan *zwdaemonv1.GetFileRequest, 1)
+	service := &testService{
+		getFile: func(
+			request *zwdaemonv1.GetFileRequest,
+			stream grpc.ServerStreamingServer[zwdaemonv1.GetFileResponse],
+		) error {
+			requests <- request
+			for _, chunk := range [][]byte{[]byte("abc"), nil, []byte("defgh")} {
+				if err := stream.Send(&zwdaemonv1.GetFileResponse{Data: chunk}); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	reader, err := client.GetFile(ctx, "folder/file.bin")
+	if err != nil {
+		t.Fatalf("GetFile() error = %v", err)
+	}
+	t.Cleanup(func() { _ = reader.Close() })
+
+	request := <-requests
+	if request.GetLogicalPath() != "folder/file.bin" {
+		t.Errorf("GetFile logical path = %q", request.GetLogicalPath())
+	}
+
+	buffer := make([]byte, 2)
+	var contents []byte
+	for {
+		n, err := reader.Read(buffer)
+		contents = append(contents, buffer[:n]...)
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			t.Fatalf("Read() error = %v", err)
+		}
+		if n == 0 {
+			t.Fatal("Read() returned no bytes and no error")
+		}
+	}
+	if got, want := string(contents), "abcdefgh"; got != want {
+		t.Fatalf("download contents = %q, want %q", got, want)
+	}
+}
+
+func TestGetFileRangeMapsRequestAndReadsResponse(t *testing.T) {
+	requests := make(chan *zwdaemonv1.GetFileRangeRequest, 1)
+	service := &testService{
+		getFileRange: func(
+			request *zwdaemonv1.GetFileRangeRequest,
+			stream grpc.ServerStreamingServer[zwdaemonv1.GetFileRangeResponse],
+		) error {
+			requests <- request
+			return stream.Send(&zwdaemonv1.GetFileRangeResponse{Data: []byte("range")})
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	reader, err := client.GetFileRange(ctx, "large.bin", 10, 15)
+	if err != nil {
+		t.Fatalf("GetFileRange() error = %v", err)
+	}
+	t.Cleanup(func() { _ = reader.Close() })
+
+	contents, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatalf("read range: %v", err)
+	}
+	if got, want := string(contents), "range"; got != want {
+		t.Fatalf("range contents = %q, want %q", got, want)
+	}
+
+	request := <-requests
+	if request.GetLogicalPath() != "large.bin" || request.GetStart() != 10 || request.GetExclusiveEnd() != 15 {
+		t.Fatalf("GetFileRange request = %#v", request)
+	}
+}
+
+func TestGetFileReturnsTerminalStructuredErrorAfterReceivedData(t *testing.T) {
+	serverStatus, err := status.New(codes.DataLoss, "download failed").WithDetails(
+		&zwdaemonv1.ZwErrorDetail{
+			Code:      zwdaemonv1.ZwErrorCode_ZW_ERROR_CODE_INTERNAL,
+			Retryable: true,
+		},
+	)
+	if err != nil {
+		t.Fatalf("create structured status: %v", err)
+	}
+	service := &testService{
+		getFile: func(
+			_ *zwdaemonv1.GetFileRequest,
+			stream grpc.ServerStreamingServer[zwdaemonv1.GetFileResponse],
+		) error {
+			if err := stream.Send(&zwdaemonv1.GetFileResponse{Data: []byte("partial")}); err != nil {
+				return err
+			}
+			return serverStatus.Err()
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	reader, err := client.GetFile(ctx, "damaged.bin")
+	if err != nil {
+		t.Fatalf("GetFile() error = %v", err)
+	}
+	t.Cleanup(func() { _ = reader.Close() })
+
+	buffer := make([]byte, 32)
+	n, err := reader.Read(buffer)
+	if err != nil {
+		t.Fatalf("first Read() error = %v", err)
+	}
+	if got, want := string(buffer[:n]), "partial"; got != want {
+		t.Fatalf("first Read() = %q, want %q", got, want)
+	}
+
+	n, err = reader.Read(buffer)
+	if n != 0 {
+		t.Errorf("terminal Read() returned %d bytes", n)
+	}
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.GRPCCode != codes.DataLoss {
+		t.Errorf("GRPCCode = %v, want %v", rpcErr.GRPCCode, codes.DataLoss)
+	}
+	if rpcErr.ZwCode == nil || *rpcErr.ZwCode != ErrorCodeInternal {
+		t.Errorf("ZwCode = %v, want %v", rpcErr.ZwCode, ErrorCodeInternal)
+	}
+	if rpcErr.Message != "download failed" || !rpcErr.Retryable {
+		t.Errorf("terminal RPC error = %#v", rpcErr)
+	}
+
+	infoCtx, infoCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer infoCancel()
+	if _, err := client.ServerInfo(infoCtx); err != nil {
+		t.Fatalf("ServerInfo() after failed download = %v", err)
+	}
+}
+
+func TestClosingDownloadCancelsStreamAndKeepsClientReusable(t *testing.T) {
+	started := make(chan struct{}, 1)
+	canceled := make(chan struct{}, 1)
+	service := &testService{
+		getFile: func(
+			_ *zwdaemonv1.GetFileRequest,
+			stream grpc.ServerStreamingServer[zwdaemonv1.GetFileResponse],
+		) error {
+			started <- struct{}{}
+			<-stream.Context().Done()
+			canceled <- struct{}{}
+			return stream.Context().Err()
+		},
+	}
+	client := startConnectedTestClient(t, service)
+
+	reader, err := client.GetFile(context.Background(), "abandoned.bin")
+	if err != nil {
+		t.Fatalf("GetFile() error = %v", err)
+	}
+	select {
+	case <-started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("server did not start download")
+	}
+
+	if err := reader.Close(); err != nil {
+		t.Fatalf("Close() error = %v", err)
+	}
+	if err := reader.Close(); err != nil {
+		t.Fatalf("second Close() error = %v", err)
+	}
+	if _, err := reader.Read(make([]byte, 1)); !errors.Is(err, fs.ErrClosed) {
+		t.Fatalf("Read() after Close() error = %v, want fs.ErrClosed", err)
+	}
+
+	select {
+	case <-canceled:
+	case <-time.After(5 * time.Second):
+		t.Fatal("closing reader did not cancel server stream")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if _, err := client.ServerInfo(ctx); err != nil {
+		t.Fatalf("ServerInfo() after closing download = %v", err)
+	}
+}
+
+func TestAddFileSendsHeaderBeforeEmptyAndMultiChunkBodies(t *testing.T) {
+	tests := []struct {
+		name       string
+		contents   []byte
+		chunkSizes []int
+	}{
+		{name: "empty"},
+		{
+			name:       "multiple_chunks",
+			contents:   bytes.Repeat([]byte("x"), 512*1024+137),
+			chunkSizes: []int{512 * 1024, 137},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			received := make(chan receivedUpload, 1)
+			service := &testService{
+				addFile: func(
+					stream grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse],
+				) error {
+					upload, err := receiveUpload(stream)
+					if err != nil {
+						return err
+					}
+					received <- upload
+					return stream.SendAndClose(&zwdaemonv1.AddFileResponse{})
+				},
+			}
+			client := startConnectedTestClient(t, service)
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+
+			err := client.AddFile(ctx, "upload.bin", bytes.NewReader(test.contents), uint64(len(test.contents)))
+			if err != nil {
+				t.Fatalf("AddFile() error = %v", err)
+			}
+
+			upload := <-received
+			if upload.header.GetLogicalPath() != "upload.bin" {
+				t.Errorf("logical path = %q", upload.header.GetLogicalPath())
+			}
+			if upload.header.GetSourceLen() != uint64(len(test.contents)) {
+				t.Errorf("source length = %d", upload.header.GetSourceLen())
+			}
+			if !bytes.Equal(upload.contents, test.contents) {
+				t.Errorf("uploaded contents differ")
+			}
+			if len(upload.chunkSizes) != len(test.chunkSizes) {
+				t.Fatalf("chunk sizes = %v, want %v", upload.chunkSizes, test.chunkSizes)
+			}
+			for index := range test.chunkSizes {
+				if upload.chunkSizes[index] != test.chunkSizes[index] {
+					t.Errorf("chunk sizes = %v, want %v", upload.chunkSizes, test.chunkSizes)
+				}
+			}
+		})
+	}
+}
+
+func TestAddFileProcessesDataReturnedWithEOF(t *testing.T) {
+	received := make(chan receivedUpload, 1)
+	service := &testService{
+		addFile: func(
+			stream grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse],
+		) error {
+			upload, err := receiveUpload(stream)
+			if err != nil {
+				return err
+			}
+			received <- upload
+			return stream.SendAndClose(&zwdaemonv1.AddFileResponse{})
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err := client.AddFile(ctx, "combined.bin", &dataAndEOFReader{data: []byte("contents")}, 8)
+	if err != nil {
+		t.Fatalf("AddFile() error = %v", err)
+	}
+	upload := <-received
+	if got, want := string(upload.contents), "contents"; got != want {
+		t.Fatalf("uploaded contents = %q, want %q", got, want)
+	}
+}
+
+func TestAddFileReturnsLengthRejectionFromCloseAndRecv(t *testing.T) {
+	serverStatus, err := status.New(codes.InvalidArgument, "declared length does not match source").WithDetails(
+		&zwdaemonv1.ZwErrorDetail{
+			Code: zwdaemonv1.ZwErrorCode_ZW_ERROR_CODE_INVALID_REQUEST,
+		},
+	)
+	if err != nil {
+		t.Fatalf("create structured status: %v", err)
+	}
+	service := &testService{
+		addFile: func(
+			stream grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse],
+		) error {
+			if _, err := receiveUpload(stream); err != nil {
+				return err
+			}
+			return serverStatus.Err()
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err = client.AddFile(ctx, "short.bin", bytes.NewReader([]byte("abc")), 4)
+	rpcErr := requireRPCError(t, err)
+	if rpcErr.GRPCCode != codes.InvalidArgument {
+		t.Errorf("GRPCCode = %v, want %v", rpcErr.GRPCCode, codes.InvalidArgument)
+	}
+	if rpcErr.ZwCode == nil || *rpcErr.ZwCode != ErrorCodeInvalidRequest {
+		t.Errorf("ZwCode = %v, want %v", rpcErr.ZwCode, ErrorCodeInvalidRequest)
+	}
+}
+
+func TestUploadReadFailureCancelsRPCAndConnectionAndPathRemainReusable(t *testing.T) {
+	injectedErr := errors.New("injected source error")
+	firstStreamErr := make(chan error, 1)
+	headers := make(chan *zwdaemonv1.AddFileHeader, 2)
+	var calls atomic.Int32
+
+	service := &testService{
+		addFile: func(
+			stream grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse],
+		) error {
+			call := calls.Add(1)
+			if call == 1 {
+				first, err := stream.Recv()
+				if err != nil {
+					firstStreamErr <- err
+					return err
+				}
+				headers <- first.GetHeader()
+
+				for {
+					_, err := stream.Recv()
+					if err != nil {
+						firstStreamErr <- err
+						return err
+					}
+				}
+			}
+
+			upload, err := receiveUpload(stream)
+			if err != nil {
+				return err
+			}
+			headers <- upload.header
+			return stream.SendAndClose(&zwdaemonv1.AddFileResponse{})
+		},
+	}
+	client := startConnectedTestClient(t, service)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err := client.AddFile(ctx, "reusable.bin", &dataAndErrorReader{
+		data: []byte("partial"),
+		err:  injectedErr,
+	}, 7)
+	var readErr *UploadReadError
+	if !errors.As(err, &readErr) {
+		t.Fatalf("AddFile() error = %T %v, want *UploadReadError", err, err)
+	}
+	if !errors.Is(err, injectedErr) {
+		t.Fatalf("AddFile() error does not wrap source error: %v", err)
+	}
+
+	select {
+	case err := <-firstStreamErr:
+		if errors.Is(err, io.EOF) {
+			t.Fatal("reader failure cleanly closed the upload stream")
+		}
+		if !errors.Is(err, context.Canceled) && status.Code(err) != codes.Canceled {
+			t.Fatalf("server stream error = %v, want cancellation", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("server did not observe upload cancellation")
+	}
+
+	infoCtx, infoCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer infoCancel()
+	if _, err := client.ServerInfo(infoCtx); err != nil {
+		t.Fatalf("ServerInfo() after reader failure = %v", err)
+	}
+
+	if err := client.AddFile(ctx, "reusable.bin", bytes.NewReader([]byte("replacement")), 11); err != nil {
+		t.Fatalf("second AddFile() on same path = %v", err)
+	}
+	for index := 0; index < 2; index++ {
+		header := <-headers
+		if header == nil || header.GetLogicalPath() != "reusable.bin" {
+			t.Fatalf("upload header = %#v", header)
+		}
+	}
+}
+
+type receivedUpload struct {
+	header     *zwdaemonv1.AddFileHeader
+	contents   []byte
+	chunkSizes []int
+}
+
+func receiveUpload(
+	stream grpc.ClientStreamingServer[zwdaemonv1.AddFileRequest, zwdaemonv1.AddFileResponse],
+) (receivedUpload, error) {
+	first, err := stream.Recv()
+	if err != nil {
+		return receivedUpload{}, err
+	}
+	header := first.GetHeader()
+	if header == nil {
+		return receivedUpload{}, status.Error(codes.InvalidArgument, "first upload message is not a header")
+	}
+
+	upload := receivedUpload{header: header}
+	for {
+		request, err := stream.Recv()
+		if err == io.EOF {
+			return upload, nil
+		}
+		if err != nil {
+			return upload, err
+		}
+		payload, ok := request.GetPayload().(*zwdaemonv1.AddFileRequest_Chunk)
+		if !ok {
+			return upload, status.Error(codes.InvalidArgument, "upload message after header is not a chunk")
+		}
+		upload.contents = append(upload.contents, payload.Chunk...)
+		upload.chunkSizes = append(upload.chunkSizes, len(payload.Chunk))
+	}
+}
+
+type dataAndEOFReader struct {
+	data []byte
+	done bool
+}
+
+func (r *dataAndEOFReader) Read(destination []byte) (int, error) {
+	if r.done {
+		return 0, io.EOF
+	}
+	r.done = true
+	return copy(destination, r.data), io.EOF
+}
+
+type dataAndErrorReader struct {
+	data []byte
+	err  error
+	done bool
+}
+
+func (r *dataAndErrorReader) Read(destination []byte) (int, error) {
+	if r.done {
+		return 0, r.err
+	}
+	r.done = true
+	return copy(destination, r.data), r.err
+}
+
+func requireRPCError(t *testing.T, err error) *RPCError {
+	t.Helper()
+
+	var rpcErr *RPCError
+	if !errors.As(err, &rpcErr) {
+		t.Fatalf("error = %T %v, want *RPCError", err, err)
+	}
+	return rpcErr
+}
+
+func statusWithRawDetail(code codes.Code, message, typeURL string, value []byte) error {
+	statusProto := status.New(code, message).Proto()
+	statusProto.Details = append(statusProto.Details, &anypb.Any{
+		TypeUrl: typeURL,
+		Value:   value,
+	})
+	return status.FromProto(statusProto).Err()
+}