Merge pull request 'fix(backup): stream volume archives through attach instead of logs' (#24) from fix/volume-backup-binary-stream into main

commit: cbddb67b36e609d5551916146616d56133a9cd75
author: bnema · 5d ago · 2026-10-17 04:52 UTC
merge of: b164fe9 + 8155d4c · diff against the first parent
signature: SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown

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	+	}