Compare commits
4
Commits
061ecc69dc
...
927fc05cfa
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
927fc05cfa | ||
|
|
f4ab1ba5ae | ||
|
|
005927a7fc | ||
|
|
09a0c01c43 |
@@ -3,6 +3,7 @@ package database
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -60,22 +61,24 @@ func (r *Repositories) WithTx(ctx context.Context, fn TxFunc) error {
|
|||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if p := recover(); p != nil {
|
if p := recover(); p != nil {
|
||||||
rollbackErr := tx.Rollback()
|
// The panic is the failure to report, so a rollback error
|
||||||
if rollbackErr != nil {
|
// is dropped.
|
||||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
_ = tx.Rollback()
|
||||||
}
|
|
||||||
|
|
||||||
panic(p)
|
panic(p)
|
||||||
} else if err != nil {
|
|
||||||
rollbackErr := tx.Rollback()
|
|
||||||
if rollbackErr != nil {
|
|
||||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
err = fn(ctx, tx)
|
err = fn(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// Once ctx is cancelled, database/sql rolls the transaction
|
||||||
|
// back itself and Rollback returns sql.ErrTxDone.
|
||||||
|
rollbackErr := tx.Rollback()
|
||||||
|
if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
|
||||||
|
return fmt.Errorf("%w; rolling back transaction: %w",
|
||||||
|
err, rollbackErr)
|
||||||
|
}
|
||||||
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
"errors"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -235,6 +236,125 @@ func TestRepositoriesTransactionRollback(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// When the context is cancelled during the body, database/sql rolls the
|
||||||
|
// transaction back itself, so WithTx's own rollback finds it already
|
||||||
|
// done. WithTx must return the body's error alone, without that rollback.
|
||||||
|
func TestRepositoriesTransactionContextCancelled(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
err := repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
file := &database.File{
|
||||||
|
Path: "/test/cancelled_file.txt",
|
||||||
|
MTime: time.Now().Truncate(time.Second),
|
||||||
|
Size: 1024,
|
||||||
|
Mode: 0644,
|
||||||
|
UID: 1000,
|
||||||
|
GID: 1000,
|
||||||
|
}
|
||||||
|
|
||||||
|
err := repos.Files.Create(ctx, tx, file)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
// database/sql rolls back from a goroutine of its own; wait
|
||||||
|
// until it has, so the test does not depend on that race.
|
||||||
|
for {
|
||||||
|
_, err = tx.ExecContext(context.WithoutCancel(ctx), "SELECT 1")
|
||||||
|
if errors.Is(err, sql.ErrTxDone) {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(time.Millisecond)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("expected context.Canceled, got: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if errors.Is(err, sql.ErrTxDone) {
|
||||||
|
t.Fatalf("the already-done rollback was reported: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
file, err := repos.Files.GetByPath(
|
||||||
|
context.Background(), "/test/cancelled_file.txt",
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("error checking for file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if file != nil {
|
||||||
|
t.Error("file should not exist after the cancelled transaction")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A body that ends the transaction itself with ROLLBACK makes WithTx's
|
||||||
|
// own rollback fail. WithTx must return that failure together with the
|
||||||
|
// body's error.
|
||||||
|
func TestRepositoriesTransactionRollbackFails(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
err := repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
_, err := tx.ExecContext(ctx, "ROLLBACK")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return errIntentionalRollback
|
||||||
|
})
|
||||||
|
if !errors.Is(err, errIntentionalRollback) {
|
||||||
|
t.Fatalf("expected the body's error, got: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !strings.Contains(err.Error(), "rolling back transaction") {
|
||||||
|
t.Errorf("expected the failed rollback in the error, got: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// When the body ends the transaction itself with ROLLBACK and then
|
||||||
|
// panics, WithTx's own rollback fails; the panic must still reach the
|
||||||
|
// caller.
|
||||||
|
func TestRepositoriesTransactionPanicRollbackFails(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
if recover() == nil {
|
||||||
|
t.Error("expected the body's panic to reach the caller")
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
_ = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
_, err := tx.ExecContext(ctx, "ROLLBACK")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
panic(errIntentionalRollback)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestRepositoriesReadTransaction(t *testing.T) {
|
func TestRepositoriesReadTransaction(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -10,6 +10,8 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
"unicode"
|
"unicode"
|
||||||
|
|
||||||
|
"github.com/dustin/go-humanize"
|
||||||
)
|
)
|
||||||
|
|
||||||
// groupSeparator joins an open group path to an attribute key. This
|
// groupSeparator joins an open group path to an attribute key. This
|
||||||
@@ -304,18 +306,13 @@ func formatDuration(d time.Duration) string {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// formatBytes formats bytes in a human-readable way
|
// formatBytes formats a byte count with humanize.Bytes, as every command
|
||||||
|
// prints one. humanize.Bytes takes no negative count, so one is printed
|
||||||
|
// as it is.
|
||||||
func formatBytes(b int64) string {
|
func formatBytes(b int64) string {
|
||||||
const unit = 1024
|
if b < 0 {
|
||||||
if b < unit {
|
|
||||||
return fmt.Sprintf("%d B", b)
|
return fmt.Sprintf("%d B", b)
|
||||||
}
|
}
|
||||||
|
|
||||||
div, exp := int64(unit), 0
|
return humanize.Bytes(uint64(b))
|
||||||
for n := b / unit; n >= unit; n /= unit {
|
|
||||||
div *= unit
|
|
||||||
exp++
|
|
||||||
}
|
|
||||||
|
|
||||||
return fmt.Sprintf("%.1f %cB", float64(b)/float64(div), "KMGTPE"[exp])
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -200,7 +200,8 @@ func TestTTYHandlerWithGroupQualifiesKeys(t *testing.T) {
|
|||||||
func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
|
func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
const oneAndAHalfKiB = 1536
|
// Printed as "1.0 MB", in powers of 1000 like every other byte count.
|
||||||
|
const oneMegabyte = 1_000_000
|
||||||
|
|
||||||
for name, testCase := range map[string]struct {
|
for name, testCase := range map[string]struct {
|
||||||
derive func(*slog.Logger) *slog.Logger
|
derive func(*slog.Logger) *slog.Logger
|
||||||
@@ -223,13 +224,13 @@ func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
|
|||||||
var buf bytes.Buffer
|
var buf bytes.Buffer
|
||||||
|
|
||||||
logger := slog.New(log.NewTTYHandler(&buf, debugHandlerOptions()))
|
logger := slog.New(log.NewTTYHandler(&buf, debugHandlerOptions()))
|
||||||
testCase.derive(logger).Info("uploaded", "bytes", oneAndAHalfKiB)
|
testCase.derive(logger).Info("uploaded", "bytes", oneMegabyte)
|
||||||
|
|
||||||
line := ansiEscape.ReplaceAllString(buf.String(), "")
|
line := ansiEscape.ReplaceAllString(buf.String(), "")
|
||||||
|
|
||||||
assert.Contains(t, line, testCase.key+"=1.5 KB",
|
assert.Contains(t, line, testCase.key+"=1.0 MB",
|
||||||
"a byte count must be human-readable however it is qualified")
|
"a byte count must be human-readable however it is qualified")
|
||||||
assert.NotContains(t, line, strconv.Itoa(oneAndAHalfKiB),
|
assert.NotContains(t, line, strconv.Itoa(oneMegabyte),
|
||||||
"the raw number must not survive the formatting")
|
"the raw number must not survive the formatting")
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
+32
-6
@@ -29,7 +29,10 @@ var (
|
|||||||
ErrURLCredentials = errors.New(
|
ErrURLCredentials = errors.New(
|
||||||
"storage URL must not carry credentials; " +
|
"storage URL must not carry credentials; " +
|
||||||
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
||||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||||
|
ErrRcloneRemoteParams = errors.New(
|
||||||
|
"rclone URL must not carry parameters in the remote name; " +
|
||||||
|
"set them on the remote with rclone config instead")
|
||||||
)
|
)
|
||||||
|
|
||||||
// URL represents a parsed storage URL.
|
// URL represents a parsed storage URL.
|
||||||
@@ -90,8 +93,9 @@ func parseFileURL(rawPath string) (*URL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||||
// the userinfo and any query parameter other than endpoint, region and
|
// the userinfo or in the endpoint parameter's userinfo, and any query
|
||||||
// ssl, so a credential-bearing URL is never stored or echoed.
|
// parameter other than endpoint, region and ssl, so a credential-bearing
|
||||||
|
// URL is never stored or echoed.
|
||||||
func parseS3URL(rawURL string) (*URL, error) {
|
func parseS3URL(rawURL string) (*URL, error) {
|
||||||
u, err := url.Parse(rawURL)
|
u, err := url.Parse(rawURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -114,6 +118,14 @@ func parseS3URL(rawURL string) (*URL, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A host name never contains "@", so one in an endpoint, with or
|
||||||
|
// without a scheme, can only introduce userinfo.
|
||||||
|
for _, endpoint := range query["endpoint"] {
|
||||||
|
if strings.Contains(endpoint, "@") {
|
||||||
|
return nil, ErrURLCredentials
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
return &URL{
|
return &URL{
|
||||||
Scheme: schemeS3,
|
Scheme: schemeS3,
|
||||||
Bucket: bucket,
|
Bucket: bucket,
|
||||||
@@ -125,9 +137,23 @@ func parseS3URL(rawURL string) (*URL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||||
// query parameters, so credentials in the userinfo and any parameter at
|
// query parameters, so credentials in the userinfo, parameters in the
|
||||||
// all are rejected rather than silently ignored.
|
// remote name and any query parameter at all are rejected rather than
|
||||||
|
// silently ignored.
|
||||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
func parseRcloneURL(rawURL string) (*URL, error) {
|
||||||
|
// The remote name runs up to the first "/", "?" or "#". It never
|
||||||
|
// contains a comma; one starts rclone's remote,param=value form,
|
||||||
|
// whose values can be credentials. This is checked before url.Parse,
|
||||||
|
// whose error for such a name can print part of it.
|
||||||
|
remote := strings.TrimPrefix(rawURL, "rclone://")
|
||||||
|
if end := strings.IndexAny(remote, "/?#"); end >= 0 {
|
||||||
|
remote = remote[:end]
|
||||||
|
}
|
||||||
|
|
||||||
|
if strings.Contains(remote, ",") {
|
||||||
|
return nil, ErrRcloneRemoteParams
|
||||||
|
}
|
||||||
|
|
||||||
u, err := url.Parse(rawURL)
|
u, err := url.Parse(rawURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, wrapParseError(err)
|
return nil, wrapParseError(err)
|
||||||
@@ -137,7 +163,7 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
|||||||
return nil, ErrURLCredentials
|
return nil, ErrURLCredentials
|
||||||
}
|
}
|
||||||
|
|
||||||
remote := u.Host
|
remote = u.Host
|
||||||
if remote == "" {
|
if remote == "" {
|
||||||
return nil, ErrMissingRemote
|
return nil, ErrMissingRemote
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -150,6 +150,19 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
|||||||
wantErr: storage.ErrURLCredentials,
|
wantErr: storage.ErrURLCredentials,
|
||||||
secrets: []string{key, secret, "mybucket"},
|
secrets: []string{key, secret, "mybucket"},
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
name: "s3 userinfo in endpoint",
|
||||||
|
raw: "s3://mybucket?endpoint=https://" +
|
||||||
|
key + ":" + secret + "@minio.example.com",
|
||||||
|
wantErr: storage.ErrURLCredentials,
|
||||||
|
secrets: []string{key, secret},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "s3 userinfo in endpoint without scheme",
|
||||||
|
raw: "s3://mybucket?endpoint=" + key + ":" + secret + "@minio.example.com",
|
||||||
|
wantErr: storage.ErrURLCredentials,
|
||||||
|
secrets: []string{key, secret},
|
||||||
|
},
|
||||||
{
|
{
|
||||||
name: "s3 unknown query param",
|
name: "s3 unknown query param",
|
||||||
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
||||||
@@ -203,6 +216,54 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestParseStorageURLRejectsRcloneRemoteParams checks that an rclone
|
||||||
|
// remote name carrying parameters (rclone's "remote,param=value" form,
|
||||||
|
// which can hold a credential) is rejected without echoing the value,
|
||||||
|
// including the URLs that url.Parse itself refuses.
|
||||||
|
func TestParseStorageURLRejectsRcloneRemoteParams(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
raw string
|
||||||
|
secrets []string // must not appear in the error message
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "configured remote",
|
||||||
|
raw: "rclone://gdrive,client_secret=topsecret/backups",
|
||||||
|
secrets: []string{"topsecret"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "remote defined inline",
|
||||||
|
raw: "rclone://:s3,access_key_id=KEY,secret_access_key=SECRET/bucket",
|
||||||
|
secrets: []string{"KEY", "SECRET"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "colon in a parameter value",
|
||||||
|
raw: "rclone://gdrive,client_secret=ab:cd/backups",
|
||||||
|
secrets: []string{"cd"},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
_, err := storage.ParseStorageURL(tc.raw)
|
||||||
|
if !errors.Is(err, storage.ErrRcloneRemoteParams) {
|
||||||
|
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
|
||||||
|
tc.raw, err, storage.ErrRcloneRemoteParams)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, secret := range tc.secrets {
|
||||||
|
if strings.Contains(err.Error(), secret) {
|
||||||
|
t.Errorf("error message leaked %q: %v", secret, err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
||||||
// itself fails, the wrapped error carries only the inner cause, not the
|
// itself fails, the wrapped error carries only the inner cause, not the
|
||||||
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
||||||
|
|||||||
@@ -103,6 +103,8 @@ func (v *Vaultik) FetchAndDecryptBlob(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// FetchBlob downloads a blob and returns a reader for the encrypted data.
|
// FetchBlob downloads a blob and returns a reader for the encrypted data.
|
||||||
|
// The reader is closed as soon as ctx is cancelled, which ends a read
|
||||||
|
// blocked inside it.
|
||||||
func (v *Vaultik) FetchBlob(
|
func (v *Vaultik) FetchBlob(
|
||||||
ctx context.Context, blobHash string,
|
ctx context.Context, blobHash string,
|
||||||
) (io.ReadCloser, error) {
|
) (io.ReadCloser, error) {
|
||||||
@@ -120,5 +122,47 @@ func (v *Vaultik) FetchBlob(
|
|||||||
return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err)
|
return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return rc, nil
|
return newCloseOnCancelReader(ctx, rc), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// closeOnCancelReader closes the reader it wraps when ctx is cancelled.
|
||||||
|
// Not every backend's reader stops on cancel by itself (the file
|
||||||
|
// backend's does not), and a read stalled on a slow or hung source would
|
||||||
|
// otherwise outlast the shutdown timeout, so the process would exit
|
||||||
|
// before restore or deep verify removed its decrypted temporary files.
|
||||||
|
type closeOnCancelReader struct {
|
||||||
|
io.ReadCloser
|
||||||
|
|
||||||
|
ctx context.Context //nolint:containedctx // Read reports its error
|
||||||
|
stop func() bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newCloseOnCancelReader(
|
||||||
|
ctx context.Context, rc io.ReadCloser,
|
||||||
|
) *closeOnCancelReader {
|
||||||
|
return &closeOnCancelReader{
|
||||||
|
ReadCloser: rc,
|
||||||
|
ctx: ctx,
|
||||||
|
stop: context.AfterFunc(ctx, func() { _ = rc.Close() }),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read returns the context's error, not the closed reader's, once ctx
|
||||||
|
// has been cancelled.
|
||||||
|
func (r *closeOnCancelReader) Read(p []byte) (int, error) {
|
||||||
|
n, err := r.ReadCloser.Read(p)
|
||||||
|
if err != nil && r.ctx.Err() != nil {
|
||||||
|
return n, r.ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close closes the wrapped reader, unless the cancel already closed it.
|
||||||
|
func (r *closeOnCancelReader) Close() error {
|
||||||
|
if !r.stop() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return r.ReadCloser.Close()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -89,22 +89,6 @@ type SnapshotInfo struct {
|
|||||||
RemotePresent *bool `json:"remote_present"`
|
RemotePresent *bool `json:"remote_present"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// formatBytes formats bytes in a human-readable format
|
|
||||||
func formatBytes(bytes int64) string {
|
|
||||||
const unit = 1024
|
|
||||||
if bytes < unit {
|
|
||||||
return fmt.Sprintf("%d B", bytes)
|
|
||||||
}
|
|
||||||
|
|
||||||
div, exp := int64(unit), 0
|
|
||||||
for n := bytes / unit; n >= unit; n /= unit {
|
|
||||||
div *= unit
|
|
||||||
exp++
|
|
||||||
}
|
|
||||||
|
|
||||||
return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp])
|
|
||||||
}
|
|
||||||
|
|
||||||
// parseSnapshotTimestamp extracts the timestamp from a snapshot ID
|
// parseSnapshotTimestamp extracts the timestamp from a snapshot ID
|
||||||
// Format: hostname_snapshotname_2026-01-12T14:41:15Z
|
// Format: hostname_snapshotname_2026-01-12T14:41:15Z
|
||||||
func parseSnapshotTimestamp(snapshotID string) (time.Time, error) {
|
func parseSnapshotTimestamp(snapshotID string) (time.Time, error) {
|
||||||
|
|||||||
@@ -142,14 +142,14 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// scratchEntries returns the vaultik blob-cache and snapshot-database
|
// scratchEntries returns the vaultik blob-cache and snapshot-database
|
||||||
// scratch entries currently present in dir.
|
// scratch entries (restore's and deep verify's) currently present in dir.
|
||||||
func scratchEntries(t *testing.T, dir string) []string {
|
func scratchEntries(t *testing.T, dir string) []string {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
var matches []string
|
var matches []string
|
||||||
|
|
||||||
for _, pattern := range []string{
|
for _, pattern := range []string{
|
||||||
"vaultik-blobcache-*", "vaultik-restore-*",
|
"vaultik-blobcache-*", "vaultik-restore-*", "vaultik-verify-*",
|
||||||
} {
|
} {
|
||||||
found, err := filepath.Glob(filepath.Join(dir, pattern))
|
found, err := filepath.Glob(filepath.Join(dir, pattern))
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -206,3 +206,119 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
|
|||||||
assert.NotContains(t, out.String(), srcPath,
|
assert.NotContains(t, out.String(), srcPath,
|
||||||
"the cancel was reported as a failed file")
|
"the cancel was reported as a failed file")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// stallingBlobStorer wraps a Storer so that every blob download hands
|
||||||
|
// over the first half of the blob and then stalls, ignoring the context,
|
||||||
|
// as a file:// blob on a hung USB stick or network share does. Only
|
||||||
|
// closing the returned reader ends the stall.
|
||||||
|
type stallingBlobStorer struct {
|
||||||
|
storage.Storer
|
||||||
|
|
||||||
|
once sync.Once
|
||||||
|
stalled chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
|
||||||
|
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *stallingBlobStorer) Get(
|
||||||
|
ctx context.Context, key string,
|
||||||
|
) (io.ReadCloser, error) {
|
||||||
|
rc, err := s.Storer.Get(ctx, key)
|
||||||
|
if err != nil || !strings.HasPrefix(key, "blobs/") {
|
||||||
|
return rc, err
|
||||||
|
}
|
||||||
|
|
||||||
|
blob, err := io.ReadAll(rc)
|
||||||
|
_ = rc.Close()
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// The writer is never closed, so once the first half has been read
|
||||||
|
// the next Read blocks until the reader is closed.
|
||||||
|
pr, pw := io.Pipe()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
_, _ = pw.Write(blob[:len(blob)/2])
|
||||||
|
|
||||||
|
s.once.Do(func() { close(s.stalled) })
|
||||||
|
}()
|
||||||
|
|
||||||
|
return pr, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a
|
||||||
|
// deep verify while a blob read is stalled on a source that ignores the
|
||||||
|
// context. Stopping must end that read, so the operation unwinds and
|
||||||
|
// removes its decrypted scratch files well within the shutdown timeout
|
||||||
|
// (issue #325).
|
||||||
|
//
|
||||||
|
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
|
||||||
|
// so it can assert on exactly the scratch files the operation created.
|
||||||
|
func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
||||||
|
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
||||||
|
return v.Restore(&RestoreOptions{
|
||||||
|
SnapshotID: snapshotID,
|
||||||
|
TargetDir: targetDir,
|
||||||
|
})
|
||||||
|
},
|
||||||
|
"deep verify": func(v *Vaultik, snapshotID, _ string) error {
|
||||||
|
return v.RunDeepVerify(snapshotID, &VerifyOptions{Deep: true})
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for name, op := range operations {
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
root := t.TempDir()
|
||||||
|
|
||||||
|
// Random content, so the blob stays large enough to stall
|
||||||
|
// part-way through its compressed, encrypted body.
|
||||||
|
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
|
||||||
|
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
|
||||||
|
|
||||||
|
scratch := filepath.Join(root, "scratch")
|
||||||
|
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
||||||
|
t.Setenv("TMPDIR", scratch)
|
||||||
|
|
||||||
|
stalling := newStallingBlobStorer(storer)
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
v := newRestoreVaultik(ctx, cfg, stalling, fs)
|
||||||
|
v.cancel = cancel
|
||||||
|
|
||||||
|
var opErr error
|
||||||
|
|
||||||
|
stop := v.StartOperation(func() {
|
||||||
|
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
|
||||||
|
})
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-stalling.stalled:
|
||||||
|
case <-time.After(30 * time.Second):
|
||||||
|
t.Fatal("the operation never reached a blob read")
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NotEmpty(t, scratchEntries(t, scratch),
|
||||||
|
"expected decrypted scratch files to exist mid-read")
|
||||||
|
|
||||||
|
stopCtx, stopCancel := context.WithTimeout(
|
||||||
|
context.Background(), 10*time.Second)
|
||||||
|
defer stopCancel()
|
||||||
|
|
||||||
|
require.True(t, stop(stopCtx),
|
||||||
|
"stop timed out; the stalled blob read was not interrupted")
|
||||||
|
require.ErrorIs(t, opErr, context.Canceled)
|
||||||
|
require.Empty(t, scratchEntries(t, scratch),
|
||||||
|
"decrypted scratch files remained after the interrupt")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -633,7 +633,7 @@ func (v *Vaultik) confirmAndExecutePurge(
|
|||||||
v.stdoutf(" %s (%s, %s)\n",
|
v.stdoutf(" %s (%s, %s)\n",
|
||||||
snap.ID,
|
snap.ID,
|
||||||
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
||||||
formatBytes(snap.CompressedSize))
|
ubytes(snap.CompressedSize))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -553,8 +553,8 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error {
|
|||||||
|
|
||||||
if snap.LocallyTracked {
|
if snap.LocallyTracked {
|
||||||
id = snap.ID.String()
|
id = snap.ID.String()
|
||||||
uncompressed = formatBytes(snap.UncompressedSize)
|
uncompressed = ubytes(snap.UncompressedSize)
|
||||||
newChunks = formatBytes(snap.NewChunkSize)
|
newChunks = ubytes(snap.NewChunkSize)
|
||||||
} else {
|
} else {
|
||||||
id = formatRemoteOnlyID(snap.RemoteKey)
|
id = formatRemoteOnlyID(snap.RemoteKey)
|
||||||
uncompressed = remoteOnlyCell
|
uncompressed = remoteOnlyCell
|
||||||
@@ -564,7 +564,7 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error {
|
|||||||
_, err = fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
|
_, err = fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
|
||||||
id,
|
id,
|
||||||
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
||||||
formatBytes(snap.CompressedSize),
|
ubytes(snap.CompressedSize),
|
||||||
uncompressed,
|
uncompressed,
|
||||||
newChunks)
|
newChunks)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -212,8 +212,8 @@ const (
|
|||||||
listConfiguredName = "home"
|
listConfiguredName = "home"
|
||||||
listLocalID = "testhost_home_2026-03-01T10:00:00Z"
|
listLocalID = "testhost_home_2026-03-01T10:00:00Z"
|
||||||
listRemoteID = "otherhost_media_2026-03-02T11:22:33Z"
|
listRemoteID = "otherhost_media_2026-03-02T11:22:33Z"
|
||||||
// fiveMegabytes formats as "5.0 MB" through formatBytes.
|
// fiveMegabytes is printed as "5.0 MB" by every command.
|
||||||
fiveMegabytes = 5 * 1024 * 1024
|
fiveMegabytes = 5_000_000
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestListSnapshots_RemoteWithoutSecretKey is the regression guard for
|
// TestListSnapshots_RemoteWithoutSecretKey is the regression guard for
|
||||||
@@ -306,6 +306,48 @@ func TestListSnapshots_RemoteOnlyRowRendering(t *testing.T) {
|
|||||||
"are not in the local index")
|
"are not in the local index")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestSnapshotListAndPurgePrintSizesLikeRemoteInfo checks that the
|
||||||
|
// `snapshot list` table and the purge listing print a byte count the way
|
||||||
|
// `remote info` does: in powers of 1000, as the MB label says.
|
||||||
|
func TestSnapshotListAndPurgePrintSizesLikeRemoteInfo(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const wantSize = "5.0 MB" // fiveMegabytes
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
|
||||||
|
// A remote-only snapshot: `snapshot list` and `remote info` both take
|
||||||
|
// its size from the manifest.
|
||||||
|
remoteKey := env.addRemote(t, listRemoteID,
|
||||||
|
time.Date(2026, 3, 2, 11, 22, 33, 0, time.UTC))
|
||||||
|
|
||||||
|
// A locally tracked snapshot of the same size for the purge to list.
|
||||||
|
// The purge prints the snapshot's blob size, the last argument here.
|
||||||
|
start := time.Date(2026, 3, 1, 10, 0, 0, 0, time.UTC)
|
||||||
|
env.addLocal(t, listLocalID, start)
|
||||||
|
env.addRemote(t, listLocalID, start)
|
||||||
|
require.NoError(t, env.v.Repositories.Snapshots.UpdateCounts(
|
||||||
|
context.Background(), nil, listLocalID, 0, 0, 0, 0, fiveMegabytes))
|
||||||
|
|
||||||
|
require.NoError(t, env.v.RemoteInfo(false))
|
||||||
|
assert.Contains(t, findTableRow(t, env.stdout.String(), remoteKey[:16]),
|
||||||
|
wantSize, "remote info")
|
||||||
|
|
||||||
|
env.stdout.Reset()
|
||||||
|
require.NoError(t, env.v.ListSnapshots(false))
|
||||||
|
assert.Contains(t, findTableRow(t, env.stdout.String(),
|
||||||
|
"<remote only:"+remoteKey[:12]+">"), wantSize, "snapshot list")
|
||||||
|
|
||||||
|
// Without --force and with nothing on stdin the purge lists what it
|
||||||
|
// would delete, reads no confirmation, and deletes nothing.
|
||||||
|
env.stdout.Reset()
|
||||||
|
require.NoError(t, env.v.PurgeSnapshotsWithOptions(
|
||||||
|
&vaultik.SnapshotPurgeOptions{OlderThan: "1d"}))
|
||||||
|
assert.Contains(t, findTableRow(t, env.stdout.String(), listLocalID),
|
||||||
|
wantSize, "snapshot purge")
|
||||||
|
}
|
||||||
|
|
||||||
// findTableRow returns the single output line containing needle.
|
// findTableRow returns the single output line containing needle.
|
||||||
func findTableRow(t *testing.T, out, needle string) string {
|
func findTableRow(t *testing.T, out, needle string) string {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|||||||
Reference in New Issue
Block a user