Add a Download button that exports a database target's archive as gzipped JSON (closes #374)

Each database target on the webhook page has a Download button that streams its archive as gzipped JSON, archive-WEBHOOKNAME-TARGETNAME-TIME.json.gz, with names made safe by delivery.ArchiveFileName's function. The export reads one consistent snapshot through one cursor in a read-only transaction, so archive writes carry on, and holds the rename lock only while it reads the stored names and opens the file. It extends its write deadline as it writes, so a large archive downloads for as long as the client reads; a failure after the response has started aborts the connection so the browser marks the download failed. The request limit is now the service's own middleware, which no longer writes a 504 over a response already started.

Model: opus-5-5

internal/handlers/handlers.go

@@ -97,7 +97,8 @@ type Handlers struct {
 	// names through the archive rename, the save and any move back.
 	// Interleaved, one could rename an archive between another's
 	// rename and save, leaving the file named for one edit and the
-	// stored names from the other.
+	// stored names from the other. An archive download holds it while
+	// it reads the stored names and opens the file they give.
 	renameMu sync.Mutex
 
 	// dummyVerifications counts the equivalent-cost verifications

internal/handlers/source_management_test.go

@@ -191,6 +191,7 @@ func storedRetentionDays(
 type sourceTestEnv struct {
 	handlers *handlers.Handlers
 	db       *database.Database
+	dbMgr    *database.WebhookDBManager
 	archives *recordingArchives
 	cookies  []*http.Cookie
 }
@@ -204,9 +205,11 @@ func setupSourceTest(t *testing.T) *sourceTestEnv {
 
 	var db *database.Database
 
+	var dbMgr *database.WebhookDBManager
+
 	var archives *recordingArchives
 
-	app := newTestApp(t, &h, &sess, &db, &archives)
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &archives)
 	app.RequireStart()
 
 	t.Cleanup(app.RequireStop)
@@ -214,6 +217,7 @@ func setupSourceTest(t *testing.T) *sourceTestEnv {
 	return &sourceTestEnv{
 		handlers: h,
 		db:       db,
+		dbMgr:    dbMgr,
 		archives: archives,
 		cookies: authenticatedCookies(
 			t, sess, sourceTestUserID, "sourceuser",

internal/handlers/target_download.go

@@ -0,0 +1,121 @@
+package handlers
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"time"
+
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+)
+
+// downloadWriteTimeout is how long one write of a download may wait
+// for a client that has stopped reading.
+const downloadWriteTimeout = 60 * time.Second
+
+// HandleTargetDownload serves a database target's archive as one
+// gzipped JSON file, named for the webhook, the target and the time;
+// see delivery.ArchiveExport.WriteGzipJSON for what it holds. Other
+// target types have no archive and are a 404.
+//
+// A download runs for as long as the client keeps reading: it reads
+// under a context the request limit does not cancel, and gives each
+// write its own deadline in place of the server's write timeout. It
+// stops when a write fails.
+func (h *Handlers) HandleTargetDownload() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		ctx := context.WithoutCancel(r.Context())
+
+		webhook, target, export, ok := h.openTargetArchive(ctx, w, r)
+		if !ok {
+			return
+		}
+
+		defer func() { _ = export.Close() }()
+
+		now := time.Now()
+
+		w.Header().Set("Content-Type", "application/gzip")
+		w.Header().Set(
+			"Content-Disposition",
+			`attachment; filename="`+delivery.ArchiveExportFileName(
+				webhook.Name, target.Name, now,
+			)+`"`,
+		)
+
+		err := export.WriteGzipJSON(
+			ctx,
+			downloadWriter{w: w, rc: http.NewResponseController(w)},
+			&webhook, target, now,
+		)
+		if err != nil {
+			h.log.Error(
+				"failed to export archive",
+				"target_id", target.ID,
+				"error", err,
+			)
+
+			// The 200 has gone out. Aborting the connection is what
+			// tells the client the file is incomplete.
+			panic(http.ErrAbortHandler)
+		}
+	}
+}
+
+// downloadWriter writes a download to the client, giving each write
+// downloadWriteTimeout to finish.
+type downloadWriter struct {
+	w  http.ResponseWriter
+	rc *http.ResponseController
+}
+
+func (d downloadWriter) Write(b []byte) (int, error) {
+	// A writer that has no write deadline, such as a test's recorder,
+	// answers http.ErrNotSupported and needs none extended.
+	err := d.rc.SetWriteDeadline(time.Now().Add(downloadWriteTimeout))
+	if err != nil && !errors.Is(err, http.ErrNotSupported) {
+		return 0, err
+	}
+
+	return d.w.Write(b)
+}
+
+// openTargetArchive opens the archive of the request's database target
+// for export, with its reads under ctx. It reports false once it has
+// written the response.
+//
+// It holds renameMu, which every archive rename runs under, while it
+// reads the stored names and opens the file, so the file it opens is
+// the one those names give. It lets go before the export is streamed:
+// once the file is open, a rename does not affect the export.
+func (h *Handlers) openTargetArchive(
+	ctx context.Context,
+	w http.ResponseWriter,
+	r *http.Request,
+) (database.Webhook, *database.Target, *delivery.ArchiveExport, bool) {
+	h.renameMu.Lock()
+	defer h.renameMu.Unlock()
+
+	webhook, target, ok := h.ownedTarget(w, r)
+	if !ok {
+		return database.Webhook{}, nil, nil, false
+	}
+
+	if target.Type != database.TargetTypeDatabase {
+		h.renderError(w, r, http.StatusNotFound)
+
+		return database.Webhook{}, nil, nil, false
+	}
+
+	export, err := delivery.OpenArchiveExport(
+		ctx, delivery.ArchivePath(h.dbMgr, &webhook, target), h.log,
+	)
+	if err != nil {
+		h.serverError(w, r, "failed to open archive for export", err)
+
+		return database.Webhook{}, nil, nil, false
+	}
+
+	return webhook, target, export, true
+}

internal/handlers/target_download_test.go

@@ -0,0 +1,379 @@
+package handlers_test
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"crypto/rand"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"sneak.berlin/go/webhooker/internal/config"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+	"sneak.berlin/go/webhooker/internal/middleware"
+)
+
+// errClientGone is the write failure of a client that has gone away.
+var errClientGone = errors.New("client gone")
+
+// downloadPath is the archive download route of a target.
+func downloadPath(webhookID, targetID string) string {
+	return "/hook/" + webhookID + "/targets/" + targetID + "/download"
+}
+
+// renameTarget submits the edit form renaming a target to Renamed.
+func renameTarget(
+	env *sourceTestEnv, webhookID, targetID string,
+) *httptest.ResponseRecorder {
+	form := url.Values{}
+	form.Set("name", "Renamed")
+
+	return submitTargetEdit(env, webhookID, targetID, form)
+}
+
+// TestHandleTargetDownload proves a database target's archive
+// downloads as a gzipped JSON attachment named for the webhook, the
+// target and the time, here with no archive file yet, so with no
+// rows; and that a target of another type has no download.
+func TestHandleTargetDownload(t *testing.T) {
+	t.Parallel()
+
+	env := setupSourceTest(t)
+	wh := seedWebhookWithRetention(t, env.db, 7)
+	archive := seedTarget(t, env.db, wh.ID, database.TargetTypeDatabase)
+	logTarget := seedTarget(t, env.db, wh.ID, database.TargetTypeLog)
+
+	w := serveTarget(
+		env, http.MethodGet, downloadPath(wh.ID, archive.ID), nil,
+	)
+	require.Equal(t, http.StatusOK, w.Code, w.Body.String())
+	assert.Equal(t, "application/gzip", w.Header().Get("Content-Type"))
+	assert.Regexp(t,
+		`^attachment; filename="archive-seeded-t-database-`+
+			`\d{8}T\d{6}Z\.json\.gz"$`,
+		w.Header().Get("Content-Disposition"),
+	)
+
+	zr, err := gzip.NewReader(w.Body)
+	require.NoError(t, err)
+
+	var got map[string]json.RawMessage
+
+	require.NoError(t, json.NewDecoder(zr).Decode(&got))
+	assert.JSONEq(t,
+		`{"id":"`+archive.ID+`","name":"t-database"}`,
+		string(got["target"]),
+	)
+	assert.JSONEq(t, `[]`, string(got["archived_events"]))
+
+	w = serveTarget(
+		env, http.MethodGet, downloadPath(wh.ID, logTarget.ID), nil,
+	)
+	assert.Equal(t, http.StatusNotFound, w.Code)
+}
+
+// TestHandleTargetDownload_WaitsForRename proves a download reads the
+// target's names and opens its archive under the lock a rename holds:
+// started while an edit is renaming the archive, it waits, and is
+// named for the target's new name.
+func TestHandleTargetDownload_WaitsForRename(t *testing.T) {
+	t.Parallel()
+
+	env := setupSourceTest(t)
+	wh := seedWebhookWithRetention(t, env.db, 7)
+	archive := seedTarget(t, env.db, wh.ID, database.TargetTypeDatabase)
+
+	renaming, release := env.archives.BlockNextRename()
+	edited := make(chan *httptest.ResponseRecorder, 1)
+
+	go func() {
+		edited <- renameTarget(env, wh.ID, archive.ID)
+	}()
+
+	<-renaming
+
+	downloaded := make(chan *httptest.ResponseRecorder, 1)
+
+	go func() {
+		downloaded <- serveTarget(
+			env, http.MethodGet, downloadPath(wh.ID, archive.ID), nil,
+		)
+	}()
+
+	select {
+	case <-downloaded:
+		release()
+		t.Fatal("the download did not wait for the rename")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	release()
+	require.Equal(t, http.StatusSeeOther, (<-edited).Code)
+
+	w := <-downloaded
+	require.Equal(t, http.StatusOK, w.Code)
+	assert.Contains(t,
+		w.Header().Get("Content-Disposition"), "archive-seeded-renamed-",
+	)
+}
+
+// stalledWriter is a response writer whose first write waits until
+// resume is closed, closing writing when it starts to wait.
+type stalledWriter struct {
+	*httptest.ResponseRecorder
+
+	once    sync.Once
+	writing chan struct{}
+	resume  chan struct{}
+}
+
+func (s *stalledWriter) Write(b []byte) (int, error) {
+	s.once.Do(func() {
+		close(s.writing)
+		<-s.resume
+	})
+
+	return s.ResponseRecorder.Write(b)
+}
+
+// TestHandleTargetDownload_StreamsWithoutTheLock proves a download
+// lets go of the rename lock once its archive is open: while the
+// download is stalled writing, an edit can still rename the target.
+func TestHandleTargetDownload_StreamsWithoutTheLock(t *testing.T) {
+	t.Parallel()
+
+	env := setupSourceTest(t)
+	wh := seedWebhookWithRetention(t, env.db, 7)
+	archive := seedTarget(t, env.db, wh.ID, database.TargetTypeDatabase)
+
+	req := httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, downloadPath(wh.ID, archive.ID), nil,
+	)
+	for _, c := range env.cookies {
+		req.AddCookie(c)
+	}
+
+	sw := &stalledWriter{
+		ResponseRecorder: httptest.NewRecorder(),
+		writing:          make(chan struct{}),
+		resume:           make(chan struct{}),
+	}
+	downloaded := make(chan struct{})
+
+	go func() {
+		targetRouter(env).ServeHTTP(sw, req)
+		close(downloaded)
+	}()
+
+	<-sw.writing
+
+	edited := make(chan *httptest.ResponseRecorder, 1)
+
+	go func() {
+		edited <- renameTarget(env, wh.ID, archive.ID)
+	}()
+
+	select {
+	case w := <-edited:
+		assert.Equal(t, http.StatusSeeOther, w.Code)
+	case <-time.After(10 * time.Second):
+		t.Error("the rename waited for the download")
+	}
+
+	close(sw.resume)
+	<-downloaded
+	assert.Equal(t, http.StatusOK, sw.Code)
+}
+
+// seedArchive writes rows to the archive file at path, each with a
+// body of bodySize random bytes, which do not compress. Its table has
+// only the columns the test fills; an export writes the others empty.
+func seedArchive(t *testing.T, path string, rows, bodySize int) {
+	t.Helper()
+
+	db, err := database.OpenSQLite(path, database.SQLiteModeCreate)
+	require.NoError(t, err)
+
+	defer func() { require.NoError(t, db.Close()) }()
+
+	_, err = db.ExecContext(t.Context(),
+		"CREATE TABLE archived_events (id INTEGER PRIMARY KEY, body TEXT)",
+	)
+	require.NoError(t, err)
+
+	body := make([]byte, bodySize)
+
+	for range rows {
+		_, _ = rand.Read(body)
+
+		_, err = db.ExecContext(t.Context(),
+			"INSERT INTO archived_events (body) VALUES (?)", string(body),
+		)
+		require.NoError(t, err)
+	}
+}
+
+// limitedServer serves the target routes as the server does, behind the
+// access log, whose lines it returns, and the request limit, here
+// limit, which is also its write timeout. Each connection's send buffer
+// is a few KiB, so a larger response is still being written while its
+// client is not reading.
+func limitedServer(
+	t *testing.T, env *sourceTestEnv, limit time.Duration,
+) (*httptest.Server, *bytes.Buffer) {
+	t.Helper()
+
+	const sendBuffer = 4 << 10
+
+	logBuf := new(bytes.Buffer)
+	mw := middleware.NewForTest(
+		slog.New(slog.NewJSONHandler(logBuf, nil)),
+		&config.Config{Environment: config.EnvironmentDev},
+		nil,
+	)
+
+	srv := httptest.NewUnstartedServer(
+		mw.Logging()(mw.Timeout(limit)(targetRouter(env))),
+	)
+	srv.Config.WriteTimeout = limit
+	srv.Config.ConnContext = func(
+		ctx context.Context, c net.Conn,
+	) context.Context {
+		tcp, ok := c.(*net.TCPConn)
+		if assert.True(t, ok) {
+			assert.NoError(t, tcp.SetWriteBuffer(sendBuffer))
+		}
+
+		return ctx
+	}
+	srv.Start()
+	t.Cleanup(srv.Close)
+
+	return srv, logBuf
+}
+
+// TestHandleTargetDownload_OutlastsTheRequestLimit proves a download
+// runs for as long as the client keeps reading, and is logged as the
+// 200 it was. Behind a request limit and a server write timeout of a
+// tenth of a second, the client stops reading once the response has
+// started, waits three times as long, and still gets the whole file.
+// The archive is larger than the connection holds, so the download is
+// still being written while the client waits.
+func TestHandleTargetDownload_OutlastsTheRequestLimit(t *testing.T) {
+	t.Parallel()
+
+	const (
+		limit    = 100 * time.Millisecond
+		rows     = 8
+		bodySize = 64 << 10
+	)
+
+	env := setupSourceTest(t)
+	wh := seedWebhookWithRetention(t, env.db, 7)
+	archive := seedTarget(t, env.db, wh.ID, database.TargetTypeDatabase)
+	seedArchive(
+		t, delivery.ArchivePath(env.dbMgr, &wh, archive), rows, bodySize,
+	)
+
+	srv, accessLog := limitedServer(t, env, limit)
+
+	req, err := http.NewRequestWithContext(
+		t.Context(), http.MethodGet,
+		srv.URL+downloadPath(wh.ID, archive.ID), nil,
+	)
+	require.NoError(t, err)
+
+	for _, c := range env.cookies {
+		req.AddCookie(c)
+	}
+
+	resp, err := srv.Client().Do(req)
+	require.NoError(t, err)
+
+	defer func() { _ = resp.Body.Close() }()
+
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	time.Sleep(3 * limit)
+
+	zr, err := gzip.NewReader(resp.Body)
+	require.NoError(t, err)
+
+	var (
+		got    map[string]json.RawMessage
+		events []json.RawMessage
+	)
+
+	require.NoError(t, json.NewDecoder(zr).Decode(&got))
+	require.NoError(t, json.Unmarshal(got["archived_events"], &events))
+	assert.Len(t, events, rows)
+
+	// Reading to the end makes the gzip reader check that the file was
+	// finished.
+	_, err = io.ReadAll(zr)
+	require.NoError(t, err)
+
+	// Close waits for the handler, so the access log line is written.
+	srv.Close()
+
+	var access map[string]any
+
+	require.NoError(t, json.Unmarshal(accessLog.Bytes(), &access))
+	assert.EqualValues(t, http.StatusOK, access["status"])
+	assert.GreaterOrEqual(t,
+		access["latency_ms"], float64(limit.Milliseconds()),
+		"the download must outlast the request limit",
+	)
+}
+
+// brokenWriter is a response writer whose writes fail once the
+// response has started, as they do when the client goes away.
+type brokenWriter struct {
+	*httptest.ResponseRecorder
+}
+
+func (b brokenWriter) Write(p []byte) (int, error) {
+	if b.Body.Len() > 0 {
+		return 0, errClientGone
+	}
+
+	return b.ResponseRecorder.Write(p)
+}
+
+// TestHandleTargetDownload_AbortsWhenItFails proves a download that
+// fails after its response has started aborts the connection, so the
+// client sees a failed download rather than a file that looks
+// complete and does not decompress.
+func TestHandleTargetDownload_AbortsWhenItFails(t *testing.T) {
+	t.Parallel()
+
+	env := setupSourceTest(t)
+	wh := seedWebhookWithRetention(t, env.db, 7)
+	archive := seedTarget(t, env.db, wh.ID, database.TargetTypeDatabase)
+
+	req := httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, downloadPath(wh.ID, archive.ID), nil,
+	)
+	for _, c := range env.cookies {
+		req.AddCookie(c)
+	}
+
+	w := brokenWriter{ResponseRecorder: httptest.NewRecorder()}
+
+	assert.PanicsWithValue(t, http.ErrAbortHandler, func() {
+		targetRouter(env).ServeHTTP(w, req)
+	})
+	assert.Equal(t, http.StatusOK, w.Code)
+}

internal/handlers/target_edit_test.go

@@ -37,14 +37,18 @@ const (
 	editAuthHeader   = "Authorization: Bearer " + editBearerSecret
 )
 
-// targetRouter mounts the target create and edit routes on a chi
-// router so the handlers see the URL parameters they read.
+// targetRouter mounts the target create, edit and download routes on
+// a chi router so the handlers see the URL parameters they read.
 func targetRouter(env *sourceTestEnv) *chi.Mux {
 	router := chi.NewRouter()
 	router.Post(
 		"/hook/{sourceID}/targets",
 		env.handlers.HandleTargetCreate(),
 	)
+	router.Get(
+		"/hook/{sourceID}/targets/{targetID}/download",
+		env.handlers.HandleTargetDownload(),
+	)
 	router.Get(
 		"/hook/{sourceID}/targets/{targetID}/edit",
 		env.handlers.HandleTargetEdit(),