Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86ec0df2e7 |
@@ -412,8 +412,7 @@ destination store.
|
||||
entirely. Remote storage is unaffected; the next backup will do a full
|
||||
scan and re-deduplicate against existing remote blobs, and the local
|
||||
index will re-bind to the currently configured storage destination.
|
||||
Use this after changing `storage_url`, or `s3.endpoint`, `s3.bucket` or
|
||||
`s3.prefix`, to a different destination.
|
||||
Use this after changing `storage_url` to a different destination.
|
||||
* `--force`: Skip confirmation prompt
|
||||
|
||||
---
|
||||
|
||||
@@ -22,23 +22,14 @@ the tag exists and is exercised; what is left is merging `next` to
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-13: Made a failed or interrupted backup remove the temporary
|
||||
file of the blob it was packing
|
||||
([issue #302](https://git.eeqj.de/sneak/vaultik/issues/302)). When
|
||||
`snapshot create` stopped on a file error, a full temporary directory
|
||||
or Ctrl-C, that blob was never finalized and its `vaultik-blob-*.tmp`
|
||||
file, up to `blob_size_limit` of encrypted data, stayed in `$TMPDIR`;
|
||||
every such run left another. Phase 2 now removes it whenever it stops
|
||||
before the last blob is finalized.
|
||||
|
||||
- 2026-10-12: Bound the local index to the destination an `s3.*` config
|
||||
names ([issue #298](https://git.eeqj.de/sneak/vaultik/issues/298)).
|
||||
Only `storage_url` was checked, so after `s3.bucket` or `s3.prefix`
|
||||
changed, the next backup treated every chunk as already stored,
|
||||
uploaded nothing, and completed a snapshot that could not be restored.
|
||||
Without `storage_url`, the local index is now bound to `s3.endpoint`,
|
||||
`s3.bucket` and `s3.prefix`, and a change is refused as a changed
|
||||
`storage_url` is.
|
||||
- 2026-10-13: Made an S3 listing that the server cuts short an error
|
||||
([issue #297](https://git.eeqj.de/sneak/vaultik/issues/297)). A page
|
||||
marked truncated with no continuation token, as a server answering V2
|
||||
listings V1-style sends, ended the listing without an error, so
|
||||
`prune` saw only the snapshots on the first page and could delete
|
||||
blobs the rest still use. Such a page is now a listing error, and
|
||||
`prune`, `snapshot list` and `remote nuke` handle it as a destination
|
||||
that cannot be listed.
|
||||
|
||||
- 2026-10-12: Rejected a retention duration longer than about 292 years
|
||||
([issue #289](https://git.eeqj.de/sneak/vaultik/issues/289)). The
|
||||
|
||||
@@ -245,17 +245,6 @@ func (p *Packer) Flush(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DiscardCurrentBlob closes and removes the temp file of the blob being
|
||||
// packed, if any, and drops that blob. A caller that stops before Flush
|
||||
// must call it, or the temp file is left behind. Thread-safe.
|
||||
func (p *Packer) DiscardCurrentBlob() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
p.cleanupTempFile()
|
||||
p.currentBlob = nil
|
||||
}
|
||||
|
||||
// FinalizeBlob finalizes the current blob being assembled.
|
||||
// This compresses the accumulated chunks, encrypts the result, and computes
|
||||
// the content-addressed hash. The finalized blob is either passed to the
|
||||
|
||||
+29
-2
@@ -18,6 +18,13 @@ import (
|
||||
"github.com/aws/smithy-go/logging"
|
||||
)
|
||||
|
||||
// ErrNoContinuationToken is returned by ListObjects and ListObjectsStream
|
||||
// when the server marks a page of a listing truncated but sends no
|
||||
// continuation token to fetch the rest with, as a server answering V2
|
||||
// listings V1-style does.
|
||||
var ErrNoContinuationToken = errors.New(
|
||||
"server marked the listing truncated but sent no continuation token")
|
||||
|
||||
// Client wraps the AWS S3 client for vaultik operations.
|
||||
// It provides a simplified interface for S3 operations with automatic
|
||||
// prefix handling and connection management. All operations are performed
|
||||
@@ -214,7 +221,7 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
||||
})
|
||||
|
||||
for paginator.HasMorePages() {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
page, err := nextListPage(ctx, paginator)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -235,6 +242,26 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
// nextListPage fetches the next page of a listing. The SDK paginator takes
|
||||
// a page marked truncated without a continuation token for the last page
|
||||
// and stops without an error, so such a page is ErrNoContinuationToken
|
||||
// here instead.
|
||||
func nextListPage(
|
||||
ctx context.Context, paginator *s3.ListObjectsV2Paginator,
|
||||
) (*s3.ListObjectsV2Output, error) {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
truncated := aws.ToBool(page.IsTruncated)
|
||||
if truncated && aws.ToString(page.NextContinuationToken) == "" {
|
||||
return nil, ErrNoContinuationToken
|
||||
}
|
||||
|
||||
return page, nil
|
||||
}
|
||||
|
||||
// HeadObject checks if an object exists in S3.
|
||||
// Returns true if the object exists, false otherwise.
|
||||
// The key is automatically prefixed with the configured prefix.
|
||||
@@ -300,7 +327,7 @@ func (c *Client) ListObjectsStream(
|
||||
})
|
||||
|
||||
for paginator.HasMorePages() {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
page, err := nextListPage(ctx, paginator)
|
||||
if err != nil {
|
||||
ch <- ObjectInfo{Err: err}
|
||||
|
||||
|
||||
@@ -1295,11 +1295,6 @@ func (s *Scanner) batchAddFilesToSnapshot(
|
||||
func (s *Scanner) processPhase(
|
||||
ctx context.Context, filesToProcess []*FileToProcess, result *ScanResult,
|
||||
) error {
|
||||
// On an error or cancellation the blob being packed is never
|
||||
// finalized, so its temp file must be removed here. After a
|
||||
// successful flush there is no such blob and this does nothing.
|
||||
defer s.packer.DiscardCurrentBlob()
|
||||
|
||||
// Calculate total bytes to process
|
||||
var totalBytes int64
|
||||
for _, f := range filesToProcess {
|
||||
|
||||
@@ -381,60 +381,3 @@ func TestScannerCancelStopsSkipErrorsRun(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestScannerRemovesBlobTempFileWhenRunStops checks that a run stopped
|
||||
// part-way, by a file it cannot read or by a cancellation, removes the
|
||||
// temp file of the blob it was packing.
|
||||
func TestScannerRemovesBlobTempFileWhenRunStops(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// a.txt is backed up first and starts a blob; the run stops at b.txt.
|
||||
const target = "/source/b.txt"
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
newFs func(cancel context.CancelFunc) afero.Fs
|
||||
}{
|
||||
{
|
||||
name: "file cannot be read",
|
||||
newFs: func(context.CancelFunc) afero.Fs {
|
||||
return &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "cancelled",
|
||||
newFs: func(cancel context.CancelFunc) afero.Fs {
|
||||
return &cancelOnOpenFs{
|
||||
Fs: afero.NewMemMapFs(), target: target, cancel: cancel,
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
fs := tt.newFs(cancel)
|
||||
writeSkipErrorTestFile(t, fs, "/source/a.txt", "first file content")
|
||||
writeSkipErrorTestFile(t, fs, target, "second file content")
|
||||
|
||||
_, err := runSkipErrorScan(ctx, t, fs, "/source", false, nil)
|
||||
if err == nil {
|
||||
t.Fatal("expected the scan to stop with an error, got nil")
|
||||
}
|
||||
|
||||
left, err := afero.Glob(fs, filepath.Join(os.TempDir(), "vaultik-blob-*"))
|
||||
if err != nil {
|
||||
t.Fatalf("listing blob temp files: %v", err)
|
||||
}
|
||||
|
||||
if len(left) != 0 {
|
||||
t.Fatalf("expected no blob temp file left, got %v", left)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
@@ -22,13 +23,16 @@ import (
|
||||
// s3TestBucket is the bucket created for each in-process S3 server.
|
||||
const s3TestBucket = "test-bucket"
|
||||
|
||||
// Credentials for the tests that build a storer from a config.Config. The
|
||||
// in-process S3 server accepts any.
|
||||
// Credentials for the tests' S3 clients. The test servers accept any.
|
||||
const (
|
||||
s3TestAccessKeyID = "key"
|
||||
s3TestSecretAccessKey = "secret"
|
||||
)
|
||||
|
||||
// s3TestRegion is the region of the S3 clients built here with
|
||||
// s3.NewClient. The test servers ignore it.
|
||||
const s3TestRegion = "us-east-1"
|
||||
|
||||
// newS3Storer builds an s3:// backend backed by a fresh in-process
|
||||
// S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem
|
||||
// over httptest) that internal/s3 and the not-found regression test use,
|
||||
@@ -55,7 +59,7 @@ func newS3Storer(t *testing.T) storage.Storer {
|
||||
Bucket: s3TestBucket,
|
||||
AccessKeyID: "test",
|
||||
SecretAccessKey: "test",
|
||||
Region: "us-east-1",
|
||||
Region: s3TestRegion,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("new client: %v", err)
|
||||
@@ -93,6 +97,68 @@ func TestS3StorerMissingKeyMapsToErrNotFound(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3ListingCutShortIsAnError checks that List and ListStream fail on a
|
||||
// listing page marked truncated that carries no continuation token, as a
|
||||
// server answering V2 listings V1-style sends it. Taken as the last page,
|
||||
// it would hide every later key, and prune would treat the blobs of the
|
||||
// snapshots listed there as unreferenced.
|
||||
func TestS3ListingCutShortIsAnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
prefix = "metadata/"
|
||||
truncatedPage = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||
<Name>test-bucket</Name>
|
||||
<Prefix>metadata/</Prefix>
|
||||
<KeyCount>1</KeyCount>
|
||||
<MaxKeys>1</MaxKeys>
|
||||
<IsTruncated>true</IsTruncated>
|
||||
<Contents>
|
||||
<Key>metadata/snap/manifest.json.zst</Key>
|
||||
<Size>8</Size>
|
||||
</Contents>
|
||||
</ListBucketResult>`
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = io.WriteString(w, truncatedPage)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
client, err := s3.NewClient(context.Background(), s3.Config{
|
||||
Endpoint: srv.URL,
|
||||
Bucket: s3TestBucket,
|
||||
AccessKeyID: s3TestAccessKeyID,
|
||||
SecretAccessKey: s3TestSecretAccessKey,
|
||||
Region: s3TestRegion,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("new client: %v", err)
|
||||
}
|
||||
|
||||
storer := storage.NewS3Storer(client)
|
||||
ctx := context.Background()
|
||||
|
||||
keys, err := storer.List(ctx, prefix)
|
||||
if !errors.Is(err, s3.ErrNoContinuationToken) {
|
||||
t.Errorf("List = %q, %v; want ErrNoContinuationToken", keys, err)
|
||||
}
|
||||
|
||||
var streamErr error
|
||||
|
||||
for obj := range storer.ListStream(ctx, prefix) {
|
||||
if obj.Err != nil {
|
||||
streamErr = obj.Err
|
||||
}
|
||||
}
|
||||
|
||||
if !errors.Is(streamErr, s3.ErrNoContinuationToken) {
|
||||
t.Errorf("ListStream error = %v, want ErrNoContinuationToken", streamErr)
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
|
||||
// writes: the README's remote storage layout, with the prefix joined to
|
||||
// each key by one "/". s3://b/p and s3://b/p/ must be the same
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
// errStorageBindingMismatch is returned when the local index database is
|
||||
// bound to a different destination than the configured one.
|
||||
// bound to a different destination than the configured storage_url.
|
||||
var errStorageBindingMismatch = errors.New(
|
||||
"local index is bound to a different backup destination")
|
||||
|
||||
@@ -24,9 +24,8 @@ var errStorageBindingMismatch = errors.New(
|
||||
//
|
||||
// Behaviour:
|
||||
// - On first use (empty stored value), record the configured
|
||||
// destination (storage_url, or the s3.* endpoint, bucket and
|
||||
// prefix) and log the binding.
|
||||
// - When the stored value matches the configured destination, do
|
||||
// storage_url and log the binding.
|
||||
// - When the stored value matches the configured storage_url, do
|
||||
// nothing and return nil.
|
||||
// - When the two differ, refuse with an error that tells the user
|
||||
// how to recover (revert the config, or run `vaultik database
|
||||
@@ -47,10 +46,12 @@ func (v *Vaultik) EnsureStorageBinding() error {
|
||||
|
||||
configured := v.Config.StorageURL
|
||||
if configured == "" {
|
||||
// The s3.* form: bind on its endpoint, bucket and prefix,
|
||||
// written the way storage_url would name the same destination.
|
||||
s3 := v.Config.S3
|
||||
configured = "s3://" + s3.Bucket + "/" + s3.Prefix + "?endpoint=" + s3.Endpoint
|
||||
// Some legacy configs still use the split s3.* keys instead of
|
||||
// storage_url. Falling back to a synthetic URL for those would
|
||||
// hide the fact that the binding is loose. Instead, treat
|
||||
// unset as "nothing to bind against" — the check is
|
||||
// necessarily best-effort for those configs.
|
||||
return nil
|
||||
}
|
||||
|
||||
stored, err := v.Repositories.LocalMeta.Get(v.ctx, database.LocalMetaKeyStorageURL)
|
||||
@@ -95,8 +96,7 @@ func buildBindingMismatchMessage(stored, configured string) string {
|
||||
"future snapshots referencing blobs that don't exist at the new destination.\n" +
|
||||
"\n" +
|
||||
"To proceed, either:\n" +
|
||||
" - revert storage_url (or the s3.* endpoint, bucket and prefix) in your\n" +
|
||||
" config to the bound destination, or\n" +
|
||||
" - revert storage_url in your config to the bound destination, or\n" +
|
||||
" - run 'vaultik database delete' to discard the local index and rebuild it\n" +
|
||||
" from a fresh full backup against the new destination"
|
||||
}
|
||||
|
||||
@@ -76,51 +76,3 @@ func TestEnsureStorageBinding_MismatchRefuses(t *testing.T) {
|
||||
assert.Contains(t, err.Error(), "file:///mnt/backups/new")
|
||||
assert.Contains(t, err.Error(), "vaultik database delete")
|
||||
}
|
||||
|
||||
// A config with no storage_url names its destination with s3.endpoint,
|
||||
// s3.bucket and s3.prefix; a change to any of them must be refused just
|
||||
// as a changed storage_url is.
|
||||
func TestEnsureStorageBinding_S3FieldsChangeRefuses(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
bound := config.S3Config{
|
||||
Endpoint: "https://s3.example.com",
|
||||
Bucket: "bucket-a",
|
||||
Prefix: "backups",
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
changed config.S3Config
|
||||
}{
|
||||
{"bucket", config.S3Config{
|
||||
Endpoint: bound.Endpoint, Bucket: "bucket-b", Prefix: bound.Prefix,
|
||||
}},
|
||||
{"prefix", config.S3Config{
|
||||
Endpoint: bound.Endpoint, Bucket: bound.Bucket, Prefix: "other",
|
||||
}},
|
||||
{"endpoint", config.S3Config{
|
||||
Endpoint: "https://s3.other.com", Bucket: bound.Bucket, Prefix: bound.Prefix,
|
||||
}},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tv, _ := buildBindTestVaultik(t, "")
|
||||
|
||||
tv.Config.S3 = bound
|
||||
require.NoError(t, tv.EnsureStorageBinding(), "first use binds")
|
||||
require.NoError(t, tv.EnsureStorageBinding(), "same destination passes")
|
||||
|
||||
tv.Config.S3 = tt.changed
|
||||
err := tv.EnsureStorageBinding()
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(),
|
||||
"local index is bound to a different backup destination")
|
||||
assert.Contains(t, err.Error(), "vaultik database delete")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user