fix(backup): stream volume archives through attach instead of logs

Volume archives were read back from the helper container with
ContainerLogs, which goes through the Docker log driver. json-file
stores output as text and replaces invalid UTF-8, so every uploaded
archive was corrupt while the job still reported success; other
drivers cannot be read back at all.

The helper is now attached before it starts and its stdout is read
from the attach stream, byte for byte. Context cancellation closes the
hijacked connection so a stalled export cannot outlive its deadline.

Integration tests cover a binary round trip, a failing helper, and
cancellation of a stalled export.

commit: 8155d4ca40edefc03e3fe1d9532dc061040d2c03
author: brice · 5d ago · 2026-10-17 04:51 UTC
parent: b164fe9
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +185 -8

M internal/adapters/out/docker/runtime.go +36 -8


@@ -57,6 +57,20 @@ type pipeReadCloser struct {
57	57		once sync.Once
58	58		}
59	59
	60	+	// hijackedReadCloser exposes a hijacked attach connection as an io.ReadCloser.
	61	+	type hijackedReadCloser struct {
	62	+	resp client.HijackedResponse
	63	+	}
	64	+
	65	+	func (h hijackedReadCloser) Read(b []byte) (int, error) {
	66	+	return h.resp.Reader.Read(b)
	67	+	}
	68	+
	69	+	func (h hijackedReadCloser) Close() error {
	70	+	h.resp.Close()
	71	+	return nil
	72	+	}
	73	+
60	74		type removeOnCloseFile struct {
61	75		*os.File
62	76		path string

@@ -1630,11 +1644,6 @@ func (r *Runtime) ExportVolumeArchive(ctx context.Context, req domain.VolumeArch
1630	1644		if err != nil {
1631	1645		return nil, err
1632	1646		}
1633		-	if _, err := r.client.ContainerStart(ctx, created.ID, client.ContainerStartOptions{}); err != nil {
1634		-	_ = cleanup()
1635		-	return nil, log.WrapErr(err, "failed to start volume archive helper container")
1636		-	}
1637		-
1638	1647		stream, err := r.attachVolumeArchiveStream(ctx, created.ID, cleanup, log)
1639	1648		if err != nil {
1640	1649		_ = cleanup()

@@ -1715,16 +1724,35 @@ func volumeArchiveHostConfig(volumeName string) *container.HostConfig {
1715	1724		}
1716	1725		}
1717	1726
	1727	+	// attachVolumeArchiveStream attaches to the helper's stdout/stderr, then
	1728	+	// starts it. The archive is read from the attach stream, never from
	1729	+	// container logs: log drivers store output as text and corrupt binary
	1730	+	// data (json-file replaces invalid UTF-8), and some drivers cannot be
	1731	+	// read back at all. Attaching before start guarantees no output byte is
	1732	+	// missed; waiting for not-running after start reports the exit status
	1733	+	// even when the helper has already exited.
1718	1734		func (r *Runtime) attachVolumeArchiveStream(ctx context.Context, containerID string, cleanup func() error, log zerowrap.Logger) (io.ReadCloser, error) {
1719		-	logs, err := r.client.ContainerLogs(ctx, containerID, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true})
	1735	+	attached, err := r.client.ContainerAttach(ctx, containerID, client.ContainerAttachOptions{Stream: true, Stdout: true, Stderr: true})
1720	1736		if err != nil {
1721	1737		return nil, log.WrapErr(err, "failed to attach volume archive stream")
1722	1738		}
	1739	+	if _, err := r.client.ContainerStart(ctx, containerID, client.ContainerStartOptions{}); err != nil {
	1740	+	attached.Close()
	1741	+	return nil, log.WrapErr(err, "failed to start volume archive helper container")
	1742	+	}
1723	1743		waitResult := r.client.ContainerWait(ctx, containerID, client.ContainerWaitOptions{Condition: container.WaitConditionNotRunning})
	1744	+	logs := hijackedReadCloser{resp: attached.HijackedResponse}
1724	1745		statusCh, errCh := waitResult.Result, waitResult.Error
1725	1746		pr, pw := io.Pipe()
1726		-	go streamVolumeArchiveLogs(logs, pw, statusCh, errCh, cleanup, log)
1727		-	return &pipeReadCloser{pr: pr, pw: pw, original: logs, cleanup: cleanup}, nil
	1747	+	stream := &pipeReadCloser{pr: pr, pw: pw, original: logs, cleanup: cleanup}
	1748	+	// A hijacked connection ignores ctx, so cancellation must close the
	1749	+	// stream explicitly to unblock the copy and remove the helper.
	1750	+	stopCancelWatch := context.AfterFunc(ctx, func() { _ = stream.Close() })
	1751	+	go func() {
	1752	+	defer stopCancelWatch()
	1753	+	streamVolumeArchiveLogs(logs, pw, statusCh, errCh, cleanup, log)
	1754	+	}()
	1755	+	return stream, nil
1728	1756		}
1729	1757
1730	1758		func streamVolumeArchiveLogs(logs io.ReadCloser, pw *io.PipeWriter, statusCh <-chan container.WaitResponse, errCh <-chan error, cleanup func() error, log zerowrap.Logger) {

 	once     sync.Once
 }
+// hijackedReadCloser exposes a hijacked attach connection as an io.ReadCloser.
+type hijackedReadCloser struct {
+	resp client.HijackedResponse
+}
+func (h hijackedReadCloser) Read(b []byte) (int, error) {
+	return h.resp.Reader.Read(b)
+}
+func (h hijackedReadCloser) Close() error {
+	h.resp.Close()
+	return nil
+}
 type removeOnCloseFile struct {
 	*os.File
 	path string
 	if err != nil {
 		return nil, err
 	}
-	if _, err := r.client.ContainerStart(ctx, created.ID, client.ContainerStartOptions{}); err != nil {
-		_ = cleanup()
-		return nil, log.WrapErr(err, "failed to start volume archive helper container")
-	}
 	stream, err := r.attachVolumeArchiveStream(ctx, created.ID, cleanup, log)
 	if err != nil {
 		_ = cleanup()
 	}
 }
+// attachVolumeArchiveStream attaches to the helper's stdout/stderr, then
+// starts it. The archive is read from the attach stream, never from
+// container logs: log drivers store output as text and corrupt binary
+// data (json-file replaces invalid UTF-8), and some drivers cannot be
+// read back at all. Attaching before start guarantees no output byte is
+// missed; waiting for not-running after start reports the exit status
+// even when the helper has already exited.
 func (r *Runtime) attachVolumeArchiveStream(ctx context.Context, containerID string, cleanup func() error, log zerowrap.Logger) (io.ReadCloser, error) {
-	logs, err := r.client.ContainerLogs(ctx, containerID, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true})
+	attached, err := r.client.ContainerAttach(ctx, containerID, client.ContainerAttachOptions{Stream: true, Stdout: true, Stderr: true})
 	if err != nil {
 		return nil, log.WrapErr(err, "failed to attach volume archive stream")
 	}
+	if _, err := r.client.ContainerStart(ctx, containerID, client.ContainerStartOptions{}); err != nil {
+		attached.Close()
+		return nil, log.WrapErr(err, "failed to start volume archive helper container")
+	}
 	waitResult := r.client.ContainerWait(ctx, containerID, client.ContainerWaitOptions{Condition: container.WaitConditionNotRunning})
+	logs := hijackedReadCloser{resp: attached.HijackedResponse}
 	statusCh, errCh := waitResult.Result, waitResult.Error
 	pr, pw := io.Pipe()
-	go streamVolumeArchiveLogs(logs, pw, statusCh, errCh, cleanup, log)
+	stream := &pipeReadCloser{pr: pr, pw: pw, original: logs, cleanup: cleanup}
-	return &pipeReadCloser{pr: pr, pw: pw, original: logs, cleanup: cleanup}, nil
+	// A hijacked connection ignores ctx, so cancellation must close the
+	// stream explicitly to unblock the copy and remove the helper.
+	stopCancelWatch := context.AfterFunc(ctx, func() { _ = stream.Close() })
+	go func() {
+		defer stopCancelWatch()
+		streamVolumeArchiveLogs(logs, pw, statusCh, errCh, cleanup, log)
+	}()
+	return stream, nil
 }
 func streamVolumeArchiveLogs(logs io.ReadCloser, pw *io.PipeWriter, statusCh <-chan container.WaitResponse, errCh <-chan error, cleanup func() error, log zerowrap.Logger) {

A internal/adapters/out/docker/volume_archive_integration_test.go +149 -0

@@ -0,0 +1,149 @@
	1	+	//go:build integration
	2	+
	3	+	package docker
	4	+
	5	+	import (
	6	+	"archive/tar"
	7	+	"compress/gzip"
	8	+	"context"
	9	+	"crypto/sha256"
	10	+	"encoding/hex"
	11	+	"fmt"
	12	+	"io"
	13	+	"os/exec"
	14	+	"strings"
	15	+	"testing"
	16	+	"time"
	17	+
	18	+	"github.com/stretchr/testify/assert"
	19	+	"github.com/stretchr/testify/require"
	20	+
	21	+	"github.com/bnema/gordon/internal/domain"
	22	+	)
	23	+
	24	+	const volumeArchiveTestImage = "alpine:3.20"
	25	+
	26	+	// TestRuntime_ExportVolumeArchive_Integration_BinaryRoundTrip proves the
	27	+	// exported stream is a byte-exact archive: binary content (random data
	28	+	// that is not valid UTF-8) must survive export unchanged.
	29	+	func TestRuntime_ExportVolumeArchive_Integration_BinaryRoundTrip(t *testing.T) {
	30	+	if testing.Short() {
	31	+	t.Skip("skipping integration test in short mode")
	32	+	}
	33	+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	34	+	defer cancel()
	35	+
	36	+	volume := fmt.Sprintf("gordon-archive-it-%d", time.Now().UnixNano())
	37	+	t.Cleanup(func() { _ = exec.Command("docker", "volume", "rm", "-f", volume).Run() })
	38	+
	39	+	out, err := exec.CommandContext(ctx, "docker", "run", "--rm", "-v", volume+":/v", volumeArchiveTestImage, "sh", "-c",
	40	+	"head -c 5000000 /dev/urandom > /v/blob && sha256sum /v/blob").CombinedOutput()
	41	+	require.NoError(t, err, string(out))
	42	+	wantSum := strings.Fields(string(out))[0]
	43	+
	44	+	runtime, err := NewRuntime()
	45	+	require.NoError(t, err)
	46	+
	47	+	res, err := runtime.ExportVolumeArchive(ctx, domain.VolumeArchiveRequest{
	48	+	VolumeName: volume,
	49	+	Compression: domain.VolumeBackupCompressionGzip,
	50	+	HelperImage: volumeArchiveTestImage,
	51	+	})
	52	+	require.NoError(t, err)
	53	+	defer res.Stream.Close()
	54	+
	55	+	gz, err := gzip.NewReader(res.Stream)
	56	+	require.NoError(t, err)
	57	+	tr := tar.NewReader(gz)
	58	+
	59	+	found := false
	60	+	for {
	61	+	header, err := tr.Next()
	62	+	if err == io.EOF {
	63	+	break
	64	+	}
	65	+	require.NoError(t, err)
	66	+	if strings.TrimPrefix(header.Name, "./") != "blob" {
	67	+	continue
	68	+	}
	69	+	hash := sha256.New()
	70	+	_, err = io.Copy(hash, tr)
	71	+	require.NoError(t, err)
	72	+	assert.Equal(t, wantSum, hex.EncodeToString(hash.Sum(nil)))
	73	+	found = true
	74	+	}
	75	+	assert.True(t, found, "archive must contain the blob file")
	76	+	}
	77	+
	78	+	// TestRuntime_ExportVolumeArchive_Integration_CancelUnblocksStream proves a
	79	+	// stalled export cannot outlive its context: once the reader stops, the
	80	+	// helper blocks on its output; cancellation must then fail the stream and
	81	+	// remove the helper container.
	82	+	func TestRuntime_ExportVolumeArchive_Integration_CancelUnblocksStream(t *testing.T) {
	83	+	if testing.Short() {
	84	+	t.Skip("skipping integration test in short mode")
	85	+	}
	86	+	volume := fmt.Sprintf("gordon-archive-it-cancel-%d", time.Now().UnixNano())
	87	+	t.Cleanup(func() { _ = exec.Command("docker", "volume", "rm", "-f", volume).Run() })
	88	+	// Incompressible data far larger than the attach socket buffers.
	89	+	out, err := exec.Command("docker", "run", "--rm", "-v", volume+":/v", volumeArchiveTestImage, "sh", "-c",
	90	+	"head -c 30000000 /dev/urandom > /v/blob").CombinedOutput()
	91	+	require.NoError(t, err, string(out))
	92	+
	93	+	runtime, err := NewRuntime()
	94	+	require.NoError(t, err)
	95	+
	96	+	ctx, cancel := context.WithCancel(context.Background())
	97	+	defer cancel()
	98	+	res, err := runtime.ExportVolumeArchive(ctx, domain.VolumeArchiveRequest{
	99	+	VolumeName: volume,
	100	+	Compression: domain.VolumeBackupCompressionGzip,
	101	+	HelperImage: volumeArchiveTestImage,
	102	+	})
	103	+	require.NoError(t, err)
	104	+	defer res.Stream.Close()
	105	+
	106	+	// Read a little, then stop consuming so the helper blocks on output.
	107	+	_, err = io.ReadFull(res.Stream, make([]byte, 1024))
	108	+	require.NoError(t, err)
	109	+	cancel()
	110	+
	111	+	// Without reading further, cancellation alone must tear the export down.
	112	+	require.Eventually(t, func() bool {
	113	+	ps, psErr := exec.Command("docker", "ps", "-aq", "--filter", "volume="+volume).Output()
	114	+	return psErr == nil && strings.TrimSpace(string(ps)) == ""
	115	+	}, 30time.Second, 200time.Millisecond, "helper container must be removed after cancellation")
	116	+
	117	+	_, err = io.ReadAll(res.Stream)
	118	+	require.Error(t, err, "a canceled export must not end as a complete archive")
	119	+	}
	120	+
	121	+	// TestRuntime_ExportVolumeArchive_Integration_HelperFailure proves a helper
	122	+	// that exits non-zero fails the stream instead of producing an artifact.
	123	+	func TestRuntime_ExportVolumeArchive_Integration_HelperFailure(t *testing.T) {
	124	+	if testing.Short() {
	125	+	t.Skip("skipping integration test in short mode")
	126	+	}
	127	+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	128	+	defer cancel()
	129	+
	130	+	runtime, err := NewRuntime()
	131	+	require.NoError(t, err)
	132	+
	133	+	// Docker creates the missing volume on mount; remove it afterwards.
	134	+	volume := fmt.Sprintf("gordon-archive-it-fail-%d", time.Now().UnixNano())
	135	+	t.Cleanup(func() { _ = exec.Command("docker", "volume", "rm", "-f", volume).Run() })
	136	+
	137	+	// zstd is absent from plain alpine, so the helper exits non-zero.
	138	+	res, err := runtime.ExportVolumeArchive(ctx, domain.VolumeArchiveRequest{
	139	+	VolumeName: volume,
	140	+	Compression: domain.VolumeBackupCompressionZstd,
	141	+	HelperImage: volumeArchiveTestImage,
	142	+	})
	143	+	require.NoError(t, err)
	144	+	defer res.Stream.Close()
	145	+
	146	+	_, err = io.ReadAll(res.Stream)
	147	+	require.Error(t, err)
	148	+	assert.Contains(t, err.Error(), "volume archive helper exited with code")
	149	+	}

@@ -0,0 +1,149 @@
	1	+	//go:build integration
	2	+
	3	+	package docker
	4	+
	5	+	import (
	6	+	"archive/tar"
	7	+	"compress/gzip"
	8	+	"context"
	9	+	"crypto/sha256"
	10	+	"encoding/hex"
	11	+	"fmt"
	12	+	"io"
	13	+	"os/exec"
	14	+	"strings"
	15	+	"testing"
	16	+	"time"
	17	+
	18	+	"github.com/stretchr/testify/assert"
	19	+	"github.com/stretchr/testify/require"
	20	+
	21	+	"github.com/bnema/gordon/internal/domain"
	22	+	)
	23	+
	24	+	const volumeArchiveTestImage = "alpine:3.20"
	25	+
	26	+	// TestRuntime_ExportVolumeArchive_Integration_BinaryRoundTrip proves the
	27	+	// exported stream is a byte-exact archive: binary content (random data
	28	+	// that is not valid UTF-8) must survive export unchanged.
	29	+	func TestRuntime_ExportVolumeArchive_Integration_BinaryRoundTrip(t *testing.T) {
	30	+	if testing.Short() {
	31	+	t.Skip("skipping integration test in short mode")
	32	+	}
	33	+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	34	+	defer cancel()
	35	+
	36	+	volume := fmt.Sprintf("gordon-archive-it-%d", time.Now().UnixNano())
	37	+	t.Cleanup(func() { _ = exec.Command("docker", "volume", "rm", "-f", volume).Run() })
	38	+
	39	+	out, err := exec.CommandContext(ctx, "docker", "run", "--rm", "-v", volume+":/v", volumeArchiveTestImage, "sh", "-c",
	40	+	"head -c 5000000 /dev/urandom > /v/blob && sha256sum /v/blob").CombinedOutput()
	41	+	require.NoError(t, err, string(out))
	42	+	wantSum := strings.Fields(string(out))[0]
	43	+
	44	+	runtime, err := NewRuntime()
	45	+	require.NoError(t, err)
	46	+
	47	+	res, err := runtime.ExportVolumeArchive(ctx, domain.VolumeArchiveRequest{
	48	+	VolumeName: volume,
	49	+	Compression: domain.VolumeBackupCompressionGzip,
	50	+	HelperImage: volumeArchiveTestImage,
	51	+	})
	52	+	require.NoError(t, err)
	53	+	defer res.Stream.Close()
	54	+
	55	+	gz, err := gzip.NewReader(res.Stream)
	56	+	require.NoError(t, err)
	57	+	tr := tar.NewReader(gz)
	58	+
	59	+	found := false
	60	+	for {
	61	+	header, err := tr.Next()
	62	+	if err == io.EOF {
	63	+	break
	64	+	}
	65	+	require.NoError(t, err)
	66	+	if strings.TrimPrefix(header.Name, "./") != "blob" {
	67	+	continue
	68	+	}
	69	+	hash := sha256.New()
	70	+	_, err = io.Copy(hash, tr)
	71	+	require.NoError(t, err)
	72	+	assert.Equal(t, wantSum, hex.EncodeToString(hash.Sum(nil)))
	73	+	found = true
	74	+	}
	75	+	assert.True(t, found, "archive must contain the blob file")
	76	+	}
	77	+
	78	+	// TestRuntime_ExportVolumeArchive_Integration_CancelUnblocksStream proves a
	79	+	// stalled export cannot outlive its context: once the reader stops, the
	80	+	// helper blocks on its output; cancellation must then fail the stream and
	81	+	// remove the helper container.
	82	+	func TestRuntime_ExportVolumeArchive_Integration_CancelUnblocksStream(t *testing.T) {
	83	+	if testing.Short() {
	84	+	t.Skip("skipping integration test in short mode")
	85	+	}
	86	+	volume := fmt.Sprintf("gordon-archive-it-cancel-%d", time.Now().UnixNano())
	87	+	t.Cleanup(func() { _ = exec.Command("docker", "volume", "rm", "-f", volume).Run() })
	88	+	// Incompressible data far larger than the attach socket buffers.
	89	+	out, err := exec.Command("docker", "run", "--rm", "-v", volume+":/v", volumeArchiveTestImage, "sh", "-c",
	90	+	"head -c 30000000 /dev/urandom > /v/blob").CombinedOutput()
	91	+	require.NoError(t, err, string(out))
	92	+
	93	+	runtime, err := NewRuntime()
	94	+	require.NoError(t, err)
	95	+
	96	+	ctx, cancel := context.WithCancel(context.Background())
	97	+	defer cancel()
	98	+	res, err := runtime.ExportVolumeArchive(ctx, domain.VolumeArchiveRequest{
	99	+	VolumeName: volume,
	100	+	Compression: domain.VolumeBackupCompressionGzip,
	101	+	HelperImage: volumeArchiveTestImage,
	102	+	})
	103	+	require.NoError(t, err)
	104	+	defer res.Stream.Close()
	105	+
	106	+	// Read a little, then stop consuming so the helper blocks on output.
	107	+	_, err = io.ReadFull(res.Stream, make([]byte, 1024))
	108	+	require.NoError(t, err)
	109	+	cancel()
	110	+
	111	+	// Without reading further, cancellation alone must tear the export down.
	112	+	require.Eventually(t, func() bool {
	113	+	ps, psErr := exec.Command("docker", "ps", "-aq", "--filter", "volume="+volume).Output()
	114	+	return psErr == nil && strings.TrimSpace(string(ps)) == ""
	115	+	}, 30time.Second, 200time.Millisecond, "helper container must be removed after cancellation")
	116	+
	117	+	_, err = io.ReadAll(res.Stream)
	118	+	require.Error(t, err, "a canceled export must not end as a complete archive")
	119	+	}
	120	+
	121	+	// TestRuntime_ExportVolumeArchive_Integration_HelperFailure proves a helper
	122	+	// that exits non-zero fails the stream instead of producing an artifact.
	123	+	func TestRuntime_ExportVolumeArchive_Integration_HelperFailure(t *testing.T) {
	124	+	if testing.Short() {
	125	+	t.Skip("skipping integration test in short mode")
	126	+	}
	127	+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	128	+	defer cancel()
	129	+
	130	+	runtime, err := NewRuntime()
	131	+	require.NoError(t, err)
	132	+
	133	+	// Docker creates the missing volume on mount; remove it afterwards.
	134	+	volume := fmt.Sprintf("gordon-archive-it-fail-%d", time.Now().UnixNano())
	135	+	t.Cleanup(func() { _ = exec.Command("docker", "volume", "rm", "-f", volume).Run() })
	136	+
	137	+	// zstd is absent from plain alpine, so the helper exits non-zero.
	138	+	res, err := runtime.ExportVolumeArchive(ctx, domain.VolumeArchiveRequest{
	139	+	VolumeName: volume,
	140	+	Compression: domain.VolumeBackupCompressionZstd,
	141	+	HelperImage: volumeArchiveTestImage,
	142	+	})
	143	+	require.NoError(t, err)
	144	+	defer res.Stream.Close()
	145	+
	146	+	_, err = io.ReadAll(res.Stream)
	147	+	require.Error(t, err)
	148	+	assert.Contains(t, err.Error(), "volume archive helper exited with code")
	149	+	}