sneak/vaultik
1
0
Fork 0
Code Issues 19 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Remediate all lint findings under the canonical golangci-lint config

Browse Source 
Fix every finding surfaced by the canonical .golangci.yml with
golangci-lint v2.12.2 (refs #61), behavior-preserving throughout:

- err113: dynamic errors replaced with package-level sentinels and %w
  wrapping; direct comparisons converted to errors.Is
- goprintffuncname: printf-style helpers renamed with an f suffix
  (ui.Writer message methods, cli.ReportErrorf, database.Fatalf,
  vaultik stdoutf) and all call sites updated
- revive: stuttering type names renamed (blob.Handler, blob.WithReader,
  blob.ChunkPosition, storage.URL, storage.Info), doc comments added,
  unused parameters blanked, package comments added
- contextcheck/noctx: ctx threaded through blob.Packer
  (AddChunk/Flush/FinalizeBlob/PackChunks) and scanner call sites;
  context-aware exec and sql variants used
- funlen/cyclop/gocognit/nestif/dupl: oversized or duplicated
  functions split into focused helpers across production and test code
- paralleltest/tparallel/thelper/usetesting/testpackage: tests
  parallelized where safe (global log.Initialize kept in the serial
  phase), helpers marked, t.TempDir adopted, external test packages
  where only exported API is used
- gosec: integer conversions clamped or justified, header timeouts
  added, remaining findings suppressed with per-site justifications
- mnd/goconst/lll/wsl_v5/nlreturn/noinlineerr/errcheck and other
  mechanical findings fixed directly

Remove the deprecated log.LogOptions alias (callers migrated to
log.Options). make check is green.
This commit is contained in:
Jeffrey Paul
2026-08-07 18:51:21 +00:00
parent 6cf9211407
commit 7ae470e530
121 changed files with 8344 additions and 5406 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
internal/blob/errors.go
View File

@@ -2,5 +2,18 @@ package blob
import "errors"
// ErrBlobSizeLimitExceeded is returned when adding a chunk would exceed the blob size limit
// ErrBlobSizeLimitExceeded is returned when adding a chunk would exceed
// the blob size limit.
var ErrBlobSizeLimitExceeded = errors.New("adding chunk would exceed blob size limit")
// ErrNoRecipients is returned when a Packer is created without any age
// recipients; blobs must always be encrypted.
var ErrNoRecipients = errors.New("recipients are required - blobs must be encrypted")
// ErrInvalidMaxBlobSize is returned when the configured maximum blob size
// is zero or negative.
var ErrInvalidMaxBlobSize = errors.New("max blob size must be positive")
// ErrNoFilesystem is returned when a Packer is created without a filesystem
// for temporary files.
var ErrNoFilesystem = errors.New("filesystem is required")

251
internal/blob/packer.go
View File

@@ -32,21 +32,28 @@ import (
"sneak.berlin/go/vaultik/internal/types"
)
// BlobHandler is a callback function invoked when a blob is finalized and ready for upload.
// The handler receives a BlobWithReader containing the blob metadata and a reader for
// the compressed and encrypted blob content. The handler is responsible for uploading
// the blob to storage and cleaning up any temporary files.
type BlobHandler func(blob *BlobWithReader) error
// Handler is a callback function invoked when a blob is finalized and
// ready for upload. The handler receives a WithReader containing the
// blob metadata and a reader for the compressed and encrypted blob content.
// The handler is responsible for uploading the blob to storage and cleaning
// up any temporary files.
type Handler func(blob *WithReader) error
// PackerConfig holds configuration for creating a Packer.
// All fields except BlobHandler are required.
type PackerConfig struct {
MaxBlobSize int64 // Maximum size of a blob before forcing finalization
CompressionLevel int // Zstd compression level (1-19, higher = better compression)
Recipients []string // Age recipients for encryption
Repositories *database.Repositories // Database repositories for tracking blob metadata
BlobHandler BlobHandler // Optional callback when blob is ready for upload
Fs afero.Fs // Filesystem for temporary files
// MaxBlobSize is the maximum size of a blob before forcing finalization.
MaxBlobSize int64
// CompressionLevel is the zstd level (1-19, higher = better compression).
CompressionLevel int
// Recipients holds the age recipients for encryption.
Recipients []string
// Repositories provides database access for tracking blob metadata.
Repositories *database.Repositories
// BlobHandler is an optional callback when a blob is ready for upload.
BlobHandler Handler
// Fs is the filesystem used for temporary files.
Fs afero.Fs
}
// PendingChunk represents a chunk waiting to be inserted into the database.
@@ -62,7 +69,7 @@ type Packer struct {
maxBlobSize int64
compressionLevel int
recipients []string // Age recipients for encryption
blobHandler BlobHandler // Called when blob is ready
blobHandler Handler // Called when blob is ready
repos *database.Repositories // For creating blob records
fs afero.Fs // Filesystem for temporary files
@@ -109,21 +116,21 @@ type FinishedBlob struct {
ID string
Hash string
Data []byte // Compressed data
Chunks []*BlobChunkRef
Chunks []*ChunkPosition
CreatedTS time.Time
Uncompressed int64
Compressed int64
}
// BlobChunkRef represents a chunk's position within a blob
type BlobChunkRef struct {
// ChunkPosition represents a chunk's position within a blob
type ChunkPosition struct {
ChunkHash string
Offset int64
Length int64
}
// BlobWithReader wraps a FinishedBlob with its data reader
type BlobWithReader struct {
// WithReader wraps a FinishedBlob with its data reader
type WithReader struct {
*FinishedBlob
Reader io.ReadSeeker
@@ -136,15 +143,15 @@ type BlobWithReader struct {
// Returns an error if required configuration fields are missing or invalid.
func NewPacker(cfg PackerConfig) (*Packer, error) {
if len(cfg.Recipients) == 0 {
return nil, errors.New("recipients are required - blobs must be encrypted")
return nil, ErrNoRecipients
}
if cfg.MaxBlobSize <= 0 {
return nil, errors.New("max blob size must be positive")
return nil, ErrInvalidMaxBlobSize
}
if cfg.Fs == nil {
return nil, errors.New("filesystem is required")
return nil, ErrNoFilesystem
}
return &Packer{
@@ -162,7 +169,7 @@ func NewPacker(cfg PackerConfig) (*Packer, error) {
// The handler is responsible for uploading the blob to storage.
// If no handler is set, finalized blobs are stored in memory and can be
// retrieved with GetFinishedBlobs().
func (p *Packer) SetBlobHandler(handler BlobHandler) {
func (p *Packer) SetBlobHandler(handler Handler) {
p.mu.Lock()
defer p.mu.Unlock()
@@ -184,13 +191,13 @@ func (p *Packer) AddPendingChunk(hash string, size int64) {
// In this case, the caller should finalize the current blob and retry.
// The chunk data is written immediately and can be garbage collected after this call.
// Thread-safe.
func (p *Packer) AddChunk(chunk *ChunkRef) error {
func (p *Packer) AddChunk(ctx context.Context, chunk *ChunkRef) error {
p.mu.Lock()
defer p.mu.Unlock()
// Initialize new blob if needed
if p.currentBlob == nil {
err := p.startNewBlob()
err := p.startNewBlob(ctx)
if err != nil {
return fmt.Errorf("starting new blob: %w", err)
}
@@ -222,12 +229,12 @@ func (p *Packer) AddChunk(chunk *ChunkRef) error {
// This should be called after all chunks have been added to ensure no data is lost.
// If a BlobHandler is set, it will be called with the finalized blob.
// Thread-safe.
func (p *Packer) Flush() error {
func (p *Packer) Flush(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
if p.currentBlob != nil && len(p.currentBlob.chunks) > 0 {
err := p.finalizeCurrentBlob()
err := p.finalizeCurrentBlob(ctx)
if err != nil {
return fmt.Errorf("finalizing blob: %w", err)
}
@@ -242,7 +249,7 @@ func (p *Packer) Flush() error {
// BlobHandler (if set) or stored internally.
// Caller must handle retrying any chunk that triggered size limit exceeded.
// Not thread-safe - caller must hold the lock.
func (p *Packer) FinalizeBlob() error {
func (p *Packer) FinalizeBlob(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
@@ -250,7 +257,7 @@ func (p *Packer) FinalizeBlob() error {
return nil
}
return p.finalizeCurrentBlob()
return p.finalizeCurrentBlob(ctx)
}
// GetFinishedBlobs returns all completed blobs and clears the internal list.
@@ -267,8 +274,33 @@ func (p *Packer) GetFinishedBlobs() []*FinishedBlob {
return blobs
}
// PackChunks is a convenience method to pack multiple chunks at once.
func (p *Packer) PackChunks(ctx context.Context, chunks []*ChunkRef) error {
for _, chunk := range chunks {
err := p.AddChunk(ctx, chunk)
if errors.Is(err, ErrBlobSizeLimitExceeded) {
// Finalize current blob and retry
err = p.FinalizeBlob(ctx)
if err != nil {
return fmt.Errorf("finalizing blob before retry: %w", err)
}
// Retry the chunk
err = p.AddChunk(ctx, chunk)
if err != nil {
return fmt.Errorf(
"adding chunk %s after finalize: %w", chunk.Hash, err)
}
} else if err != nil {
return fmt.Errorf("adding chunk %s: %w", chunk.Hash, err)
}
}
return p.Flush(ctx)
}
// startNewBlob initializes a new blob (must be called with lock held)
func (p *Packer) startNewBlob() error {
func (p *Packer) startNewBlob(ctx context.Context) error {
// Generate UUID for the blob
blobID := uuid.New().String()
@@ -280,8 +312,9 @@ func (p *Packer) startNewBlob() error {
}
blob := &database.Blob{
ID: blobIDTyped,
Hash: types.BlobHash("temp-placeholder-" + blobID), // Temporary placeholder until finalized
ID: blobIDTyped,
// Temporary placeholder hash until finalized.
Hash: types.BlobHash("temp-placeholder-" + blobID),
CreatedTS: time.Now().UTC(),
FinishedTS: nil,
UncompressedSize: 0,
@@ -289,9 +322,12 @@ func (p *Packer) startNewBlob() error {
UploadedTS: nil,
}
if err := p.repos.WithTx(context.Background(), func(ctx context.Context, tx *sql.Tx) error {
return p.repos.Blobs.Create(ctx, tx, blob)
}); err != nil {
err = p.repos.WithTx(
ctx,
func(txCtx context.Context, tx *sql.Tx) error {
return p.repos.Blobs.Create(txCtx, tx, blob)
})
if err != nil {
return fmt.Errorf("creating blob record: %w", err)
}
}
@@ -321,16 +357,19 @@ func (p *Packer) startNewBlob() error {
size: 0,
}
log.Debug("Created new blob container", "blob_id", blobID, "temp_file", tempFile.Name())
log.Debug("Created new blob container",
"blob_id", blobID, "temp_file", tempFile.Name())
return nil
}
// addChunkToCurrentBlob adds a chunk to the current blob (must be called with lock held)
// addChunkToCurrentBlob adds a chunk to the current blob (must be called
// with lock held).
func (p *Packer) addChunkToCurrentBlob(chunk *ChunkRef) error {
// Skip if chunk already in current blob
if p.currentBlob.chunkSet[chunk.Hash] {
log.Debug("Skipping duplicate chunk already in current blob", "chunk_hash", chunk.Hash)
log.Debug("Skipping duplicate chunk already in current blob",
"chunk_hash", chunk.Hash)
return nil
}
@@ -339,7 +378,8 @@ func (p *Packer) addChunkToCurrentBlob(chunk *ChunkRef) error {
offset := p.currentBlob.size
// Write to the blobgen writer (compression -> encryption -> disk)
if _, err := p.currentBlob.writer.Write(chunk.Data); err != nil {
_, err := p.currentBlob.writer.Write(chunk.Data)
if err != nil {
return fmt.Errorf("writing to blob stream: %w", err)
}
@@ -372,7 +412,7 @@ func (p *Packer) addChunkToCurrentBlob(chunk *ChunkRef) error {
}
// finalizeCurrentBlob completes the current blob (must be called with lock held)
func (p *Packer) finalizeCurrentBlob() error {
func (p *Packer) finalizeCurrentBlob(ctx context.Context) error {
if p.currentBlob == nil {
return nil
}
@@ -387,7 +427,8 @@ func (p *Packer) finalizeCurrentBlob() error {
chunksToInsert := p.pendingChunks
p.pendingChunks = nil
if err := p.commitBlobToDatabase(blobHash, finalSize, chunksToInsert); err != nil {
err = p.commitBlobToDatabase(ctx, blobHash, finalSize, chunksToInsert)
if err != nil {
return err
}
@@ -415,15 +456,18 @@ func (p *Packer) finalizeCurrentBlob() error {
return p.deliverFinishedBlob(finished, insertedChunkHashes)
}
// closeBlobWriter closes the writer, syncs to disk, and returns the blob hash and final size
// closeBlobWriter closes the writer, syncs to disk, and returns the blob
// hash and final size.
func (p *Packer) closeBlobWriter() (string, int64, error) {
if err := p.currentBlob.writer.Close(); err != nil {
err := p.currentBlob.writer.Close()
if err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("closing blobgen writer: %w", err)
}
if err := p.currentBlob.tempFile.Sync(); err != nil {
err = p.currentBlob.tempFile.Sync()
if err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("syncing temp file: %w", err)
@@ -436,7 +480,8 @@ func (p *Packer) closeBlobWriter() (string, int64, error) {
return "", 0, fmt.Errorf("getting file size: %w", err)
}
if _, err := p.currentBlob.tempFile.Seek(0, io.SeekStart); err != nil {
_, err = p.currentBlob.tempFile.Seek(0, io.SeekStart)
if err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("seeking to start: %w", err)
@@ -447,11 +492,11 @@ func (p *Packer) closeBlobWriter() (string, int64, error) {
return hex.EncodeToString(finalHash), finalSize, nil
}
// buildChunkRefs creates BlobChunkRef entries from the current blob's chunks
func (p *Packer) buildChunkRefs() []*BlobChunkRef {
refs := make([]*BlobChunkRef, 0, len(p.currentBlob.chunks))
// buildChunkRefs creates ChunkPosition entries from the current blob's chunks
func (p *Packer) buildChunkRefs() []*ChunkPosition {
refs := make([]*ChunkPosition, 0, len(p.currentBlob.chunks))
for _, chunk := range p.currentBlob.chunks {
refs = append(refs, &BlobChunkRef{
refs = append(refs, &ChunkPosition{
ChunkHash: chunk.Hash, Offset: chunk.Offset, Length: chunk.Size,
})
}
@@ -460,7 +505,10 @@ func (p *Packer) buildChunkRefs() []*BlobChunkRef {
}
// commitBlobToDatabase inserts pending chunks, blob_chunks, and updates the blob record
func (p *Packer) commitBlobToDatabase(blobHash string, finalSize int64, chunksToInsert []PendingChunk) error {
func (p *Packer) commitBlobToDatabase(
ctx context.Context,
blobHash string, finalSize int64, chunksToInsert []PendingChunk,
) error {
if p.repos == nil {
return nil
}
@@ -472,30 +520,12 @@ func (p *Packer) commitBlobToDatabase(blobHash string, finalSize int64, chunksTo
return fmt.Errorf("parsing blob ID: %w", parseErr)
}
err := p.repos.WithTx(context.Background(), func(ctx context.Context, tx *sql.Tx) error {
for _, chunk := range chunksToInsert {
dbChunk := &database.Chunk{ChunkHash: types.ChunkHash(chunk.Hash), Size: chunk.Size}
err := p.repos.Chunks.Create(ctx, tx, dbChunk)
if err != nil {
return fmt.Errorf("creating chunk: %w", err)
}
}
for _, chunk := range p.currentBlob.chunks {
blobChunk := &database.BlobChunk{
BlobID: blobIDTyped, ChunkHash: types.ChunkHash(chunk.Hash),
Offset: chunk.Offset, Length: chunk.Size,
}
err := p.repos.BlobChunks.Create(ctx, tx, blobChunk)
if err != nil {
return fmt.Errorf("creating blob_chunk: %w", err)
}
}
return p.repos.Blobs.UpdateFinished(ctx, tx, p.currentBlob.id, blobHash, p.currentBlob.size, finalSize)
})
err := p.repos.WithTx(
ctx,
func(txCtx context.Context, tx *sql.Tx) error {
return p.insertBlobRecords(txCtx, tx, blobIDTyped, blobHash,
finalSize, chunksToInsert)
})
if err != nil {
p.cleanupTempFile()
@@ -503,28 +533,69 @@ func (p *Packer) commitBlobToDatabase(blobHash string, finalSize int64, chunksTo
}
log.Debug("Committed blob transaction",
"chunks_inserted", len(chunksToInsert), "blob_chunks_inserted", len(p.currentBlob.chunks))
"chunks_inserted", len(chunksToInsert),
"blob_chunks_inserted", len(p.currentBlob.chunks))
return nil
}
// insertBlobRecords inserts pending chunks and blob_chunk rows, then marks
// the blob finished, all within the supplied transaction.
func (p *Packer) insertBlobRecords(
ctx context.Context,
tx *sql.Tx,
blobIDTyped types.BlobID,
blobHash string,
finalSize int64,
chunksToInsert []PendingChunk,
) error {
for _, chunk := range chunksToInsert {
dbChunk := &database.Chunk{
ChunkHash: types.ChunkHash(chunk.Hash), Size: chunk.Size,
}
err := p.repos.Chunks.Create(ctx, tx, dbChunk)
if err != nil {
return fmt.Errorf("creating chunk: %w", err)
}
}
for _, chunk := range p.currentBlob.chunks {
blobChunk := &database.BlobChunk{
BlobID: blobIDTyped, ChunkHash: types.ChunkHash(chunk.Hash),
Offset: chunk.Offset, Length: chunk.Size,
}
err := p.repos.BlobChunks.Create(ctx, tx, blobChunk)
if err != nil {
return fmt.Errorf("creating blob_chunk: %w", err)
}
}
return p.repos.Blobs.UpdateFinished(ctx, tx, p.currentBlob.id, blobHash,
p.currentBlob.size, finalSize)
}
// deliverFinishedBlob passes the blob to the handler or stores it internally
func (p *Packer) deliverFinishedBlob(finished *FinishedBlob, insertedChunkHashes []string) error {
func (p *Packer) deliverFinishedBlob(
finished *FinishedBlob, insertedChunkHashes []string,
) error {
if p.blobHandler != nil {
if _, err := p.currentBlob.tempFile.Seek(0, io.SeekStart); err != nil {
_, err := p.currentBlob.tempFile.Seek(0, io.SeekStart)
if err != nil {
p.cleanupTempFile()
return fmt.Errorf("seeking for handler: %w", err)
}
blobWithReader := &BlobWithReader{
blobWithReader := &WithReader{
FinishedBlob: finished,
Reader: p.currentBlob.tempFile,
TempFile: p.currentBlob.tempFile,
InsertedChunkHashes: insertedChunkHashes,
}
err := p.blobHandler(blobWithReader)
err = p.blobHandler(blobWithReader)
if err != nil {
p.cleanupTempFile()
@@ -539,7 +610,8 @@ func (p *Packer) deliverFinishedBlob(finished *FinishedBlob, insertedChunkHashes
// No handler - read data for legacy behavior
log.Debug("No blob handler callback configured", "blob_hash", finished.Hash[:8]+"...")
if _, err := p.currentBlob.tempFile.Seek(0, io.SeekStart); err != nil {
_, err := p.currentBlob.tempFile.Seek(0, io.SeekStart)
if err != nil {
p.cleanupTempFile()
return fmt.Errorf("seeking to read data: %w", err)
@@ -568,26 +640,3 @@ func (p *Packer) cleanupTempFile() {
_ = p.fs.Remove(name)
}
}
// PackChunks is a convenience method to pack multiple chunks at once
func (p *Packer) PackChunks(chunks []*ChunkRef) error {
for _, chunk := range chunks {
err := p.AddChunk(chunk)
if errors.Is(err, ErrBlobSizeLimitExceeded) {
// Finalize current blob and retry
err := p.FinalizeBlob()
if err != nil {
return fmt.Errorf("finalizing blob before retry: %w", err)
}
// Retry the chunk
err = p.AddChunk(chunk)
if err != nil {
return fmt.Errorf("adding chunk %s after finalize: %w", chunk.Hash, err)
}
} else if err != nil {
return fmt.Errorf("adding chunk %s: %w", chunk.Hash, err)
}
}
return p.Flush()
}

664
internal/blob/packer_test.go
View File

@@ -1,4 +1,4 @@
package blob
package blob_test
import (
"bytes"
@@ -13,6 +13,7 @@ import (
"filippo.io/age"
"github.com/klauspost/compress/zstd"
"github.com/spf13/afero"
"sneak.berlin/go/vaultik/internal/blob"
"sneak.berlin/go/vaultik/internal/database"
"sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/types"
@@ -20,387 +21,298 @@ import (
const (
// Test key from test/insecure-integration-test.key
testPrivateKey = "AGE-SECRET-KEY-19CR5YSFW59HM4TLD6GXVEDMZFTVVF7PPHKUT68TXSFPK7APHXA2QS2NJA5"
testPublicKey = "age1ezrjmfpwsc95svdg0y54mums3zevgzu0x0ecq2f7tp8a05gl0sjq9q9wjg"
testPrivateKey = "AGE-SECRET-KEY-19CR5YSFW59HM4TLD6GXVEDMZFTVVF7PPHKUT68TXSFPK7A" +
"PHXA2QS2NJA5"
testPublicKey = "age1ezrjmfpwsc95svdg0y54mums3zevgzu0x0ecq2f7tp8a05gl0sjq9q9wjg"
defaultMaxBlobSize = 10 * 1024 * 1024 // 10MB
testChunkSize = 1000
testChunkCount = 10
)
func TestPacker(t *testing.T) {
// Initialize logger for tests
log.Initialize(log.Config{})
// parseTestIdentity parses the fixed test age identity.
func parseTestIdentity(t *testing.T) *age.X25519Identity {
t.Helper()
// Parse test identity
identity, err := age.ParseX25519Identity(testPrivateKey)
if err != nil {
t.Fatalf("failed to parse test identity: %v", err)
}
t.Run("single chunk creates single blob", func(t *testing.T) {
// Create test database
db, err := database.NewTestDB()
if err != nil {
t.Fatalf("failed to create test db: %v", err)
}
defer func() { _ = db.Close() }()
repos := database.NewRepositories(db)
cfg := PackerConfig{
MaxBlobSize: 10 * 1024 * 1024, // 10MB
CompressionLevel: 3,
Recipients: []string{testPublicKey},
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
}
// Create a chunk
data := []byte("Hello, World!")
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
// Create chunk in database first
dbChunk := &database.Chunk{
ChunkHash: types.ChunkHash(hashStr),
Size: int64(len(data)),
}
err = repos.WithTx(context.Background(), func(ctx context.Context, tx *sql.Tx) error {
return repos.Chunks.Create(ctx, tx, dbChunk)
})
if err != nil {
t.Fatalf("failed to create chunk in db: %v", err)
}
chunk := &ChunkRef{
Hash: hashStr,
Data: data,
}
// Add chunk
if err := packer.AddChunk(chunk); err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
// Flush
if err := packer.Flush(); err != nil {
t.Fatalf("failed to flush: %v", err)
}
// Get finished blobs
blobs := packer.GetFinishedBlobs()
if len(blobs) != 1 {
t.Fatalf("expected 1 blob, got %d", len(blobs))
}
blob := blobs[0]
if len(blob.Chunks) != 1 {
t.Errorf("expected 1 chunk in blob, got %d", len(blob.Chunks))
}
// Note: Very small data may not compress well
t.Logf("Compression: %d -> %d bytes", blob.Uncompressed, blob.Compressed)
// Decrypt the blob data
decrypted, err := age.Decrypt(bytes.NewReader(blob.Data), identity)
if err != nil {
t.Fatalf("failed to decrypt blob: %v", err)
}
// Decompress the decrypted data
reader, err := zstd.NewReader(decrypted)
if err != nil {
t.Fatalf("failed to create decompressor: %v", err)
}
defer reader.Close()
var decompressed bytes.Buffer
if _, err := io.Copy(&decompressed, reader); err != nil {
t.Fatalf("failed to decompress: %v", err)
}
if !bytes.Equal(decompressed.Bytes(), data) {
t.Error("decompressed data doesn't match original")
}
})
t.Run("multiple chunks packed together", func(t *testing.T) {
// Create test database
db, err := database.NewTestDB()
if err != nil {
t.Fatalf("failed to create test db: %v", err)
}
defer func() { _ = db.Close() }()
repos := database.NewRepositories(db)
cfg := PackerConfig{
MaxBlobSize: 10 * 1024 * 1024, // 10MB
CompressionLevel: 3,
Recipients: []string{testPublicKey},
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
}
// Create multiple small chunks
chunks := make([]*ChunkRef, 10)
for i := range 10 {
data := bytes.Repeat([]byte{byte(i)}, 1000)
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
// Create chunk in database first
dbChunk := &database.Chunk{
ChunkHash: types.ChunkHash(hashStr),
Size: int64(len(data)),
}
err = repos.WithTx(context.Background(), func(ctx context.Context, tx *sql.Tx) error {
return repos.Chunks.Create(ctx, tx, dbChunk)
})
if err != nil {
t.Fatalf("failed to create chunk in db: %v", err)
}
chunks[i] = &ChunkRef{
Hash: hashStr,
Data: data,
}
}
// Add all chunks
for _, chunk := range chunks {
err := packer.AddChunk(chunk)
if err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
}
// Flush
if err := packer.Flush(); err != nil {
t.Fatalf("failed to flush: %v", err)
}
// Should have one blob with all chunks
blobs := packer.GetFinishedBlobs()
if len(blobs) != 1 {
t.Fatalf("expected 1 blob, got %d", len(blobs))
}
if len(blobs[0].Chunks) != 10 {
t.Errorf("expected 10 chunks in blob, got %d", len(blobs[0].Chunks))
}
// Verify offsets are correct
expectedOffset := int64(0)
for i, chunkRef := range blobs[0].Chunks {
if chunkRef.Offset != expectedOffset {
t.Errorf("chunk %d: expected offset %d, got %d", i, expectedOffset, chunkRef.Offset)
}
if chunkRef.Length != 1000 {
t.Errorf("chunk %d: expected length 1000, got %d", i, chunkRef.Length)
}
expectedOffset += chunkRef.Length
}
})
t.Run("blob size limit enforced", func(t *testing.T) {
// Create test database
db, err := database.NewTestDB()
if err != nil {
t.Fatalf("failed to create test db: %v", err)
}
defer func() { _ = db.Close() }()
repos := database.NewRepositories(db)
// Small blob size limit to force multiple blobs
cfg := PackerConfig{
MaxBlobSize: 5000, // 5KB max
CompressionLevel: 3,
Recipients: []string{testPublicKey},
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
}
// Create chunks that will exceed the limit
chunks := make([]*ChunkRef, 10)
for i := range 10 {
data := bytes.Repeat([]byte{byte(i)}, 1000) // 1KB each
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
// Create chunk in database first
dbChunk := &database.Chunk{
ChunkHash: types.ChunkHash(hashStr),
Size: int64(len(data)),
}
err = repos.WithTx(context.Background(), func(ctx context.Context, tx *sql.Tx) error {
return repos.Chunks.Create(ctx, tx, dbChunk)
})
if err != nil {
t.Fatalf("failed to create chunk in db: %v", err)
}
chunks[i] = &ChunkRef{
Hash: hashStr,
Data: data,
}
}
blobCount := 0
// Add chunks and handle size limit errors
for _, chunk := range chunks {
err := packer.AddChunk(chunk)
if errors.Is(err, ErrBlobSizeLimitExceeded) {
// Finalize current blob
err := packer.FinalizeBlob()
if err != nil {
t.Fatalf("failed to finalize blob: %v", err)
}
blobCount++
// Retry adding the chunk
err = packer.AddChunk(chunk)
if err != nil {
t.Fatalf("failed to add chunk after finalize: %v", err)
}
} else if err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
}
// Flush remaining
if err := packer.Flush(); err != nil {
t.Fatalf("failed to flush: %v", err)
}
// Get all blobs
blobs := packer.GetFinishedBlobs()
totalBlobs := blobCount + len(blobs)
// Should have multiple blobs due to size limit
if totalBlobs < 2 {
t.Errorf("expected multiple blobs due to size limit, got %d", totalBlobs)
}
// Verify each blob respects size limit (approximately)
for _, blob := range blobs {
if blob.Compressed > 6000 { // Allow some overhead
t.Errorf("blob size %d exceeds limit", blob.Compressed)
}
}
})
t.Run("with encryption", func(t *testing.T) {
// Create test database
db, err := database.NewTestDB()
if err != nil {
t.Fatalf("failed to create test db: %v", err)
}
defer func() { _ = db.Close() }()
repos := database.NewRepositories(db)
// Generate test identity (using the one from parent test)
cfg := PackerConfig{
MaxBlobSize: 10 * 1024 * 1024, // 10MB
CompressionLevel: 3,
Recipients: []string{testPublicKey},
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
}
// Create test data
data := bytes.Repeat([]byte("Test data for encryption!"), 100)
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
// Create chunk in database first
dbChunk := &database.Chunk{
ChunkHash: types.ChunkHash(hashStr),
Size: int64(len(data)),
}
err = repos.WithTx(context.Background(), func(ctx context.Context, tx *sql.Tx) error {
return repos.Chunks.Create(ctx, tx, dbChunk)
})
if err != nil {
t.Fatalf("failed to create chunk in db: %v", err)
}
chunk := &ChunkRef{
Hash: hashStr,
Data: data,
}
// Add chunk and flush
if err := packer.AddChunk(chunk); err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
if err := packer.Flush(); err != nil {
t.Fatalf("failed to flush: %v", err)
}
// Get blob
blobs := packer.GetFinishedBlobs()
if len(blobs) != 1 {
t.Fatalf("expected 1 blob, got %d", len(blobs))
}
blob := blobs[0]
// Decrypt the blob
decrypted, err := age.Decrypt(bytes.NewReader(blob.Data), identity)
if err != nil {
t.Fatalf("failed to decrypt blob: %v", err)
}
var decryptedData bytes.Buffer
if _, err := decryptedData.ReadFrom(decrypted); err != nil {
t.Fatalf("failed to read decrypted data: %v", err)
}
// Decompress
reader, err := zstd.NewReader(&decryptedData)
if err != nil {
t.Fatalf("failed to create decompressor: %v", err)
}
defer reader.Close()
var decompressed bytes.Buffer
if _, err := decompressed.ReadFrom(reader); err != nil {
t.Fatalf("failed to decompress: %v", err)
}
// Verify data
if !bytes.Equal(decompressed.Bytes(), data) {
t.Error("decrypted and decompressed data doesn't match original")
}
})
return identity
}
// newTestPacker creates a test database and a Packer backed by it.
func newTestPacker(
t *testing.T, maxBlobSize int64,
) (*database.Repositories, *blob.Packer) {
t.Helper()
db, err := database.NewTestDB()
if err != nil {
t.Fatalf("failed to create test db: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
repos := database.NewRepositories(db)
packer, err := blob.NewPacker(blob.PackerConfig{
MaxBlobSize: maxBlobSize,
CompressionLevel: 3,
Recipients: []string{testPublicKey},
Repositories: repos,
Fs: afero.NewMemMapFs(),
})
if err != nil {
t.Fatalf("failed to create packer: %v", err)
}
return repos, packer
}
// makeChunk creates a ChunkRef for data and registers the chunk in the
// database.
func makeChunk(
t *testing.T, repos *database.Repositories, data []byte,
) *blob.ChunkRef {
t.Helper()
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
dbChunk := &database.Chunk{
ChunkHash: types.ChunkHash(hashStr),
Size: int64(len(data)),
}
err := repos.WithTx(
context.Background(),
func(ctx context.Context, tx *sql.Tx) error {
return repos.Chunks.Create(ctx, tx, dbChunk)
})
if err != nil {
t.Fatalf("failed to create chunk in db: %v", err)
}
return &blob.ChunkRef{
Hash: hashStr,
Data: data,
}
}
// decryptAndDecompress reverses the blob pipeline: age decrypt, then zstd
// decompress.
func decryptAndDecompress(
t *testing.T, blobData []byte, identity *age.X25519Identity,
) []byte {
t.Helper()
decrypted, err := age.Decrypt(bytes.NewReader(blobData), identity)
if err != nil {
t.Fatalf("failed to decrypt blob: %v", err)
}
reader, err := zstd.NewReader(decrypted)
if err != nil {
t.Fatalf("failed to create decompressor: %v", err)
}
defer reader.Close()
var decompressed bytes.Buffer
_, err = io.Copy(&decompressed, reader)
if err != nil {
t.Fatalf("failed to decompress: %v", err)
}
return decompressed.Bytes()
}
func TestPackerSingleChunk(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
identity := parseTestIdentity(t)
repos, packer := newTestPacker(t, defaultMaxBlobSize)
ctx := context.Background()
data := []byte("Hello, World!")
chunk := makeChunk(t, repos, data)
err := packer.AddChunk(ctx, chunk)
if err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
err = packer.Flush(ctx)
if err != nil {
t.Fatalf("failed to flush: %v", err)
}
blobs := packer.GetFinishedBlobs()
if len(blobs) != 1 {
t.Fatalf("expected 1 blob, got %d", len(blobs))
}
finished := blobs[0]
if len(finished.Chunks) != 1 {
t.Errorf("expected 1 chunk in blob, got %d", len(finished.Chunks))
}
// Note: Very small data may not compress well
t.Logf("Compression: %d -> %d bytes",
finished.Uncompressed, finished.Compressed)
decompressed := decryptAndDecompress(t, finished.Data, identity)
if !bytes.Equal(decompressed, data) {
t.Error("decompressed data doesn't match original")
}
}
func TestPackerMultipleChunks(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
repos, packer := newTestPacker(t, defaultMaxBlobSize)
ctx := context.Background()
chunks := make([]*blob.ChunkRef, testChunkCount)
for i := range testChunkCount {
data := bytes.Repeat([]byte{byte(i)}, testChunkSize)
chunks[i] = makeChunk(t, repos, data)
}
for _, chunk := range chunks {
err := packer.AddChunk(ctx, chunk)
if err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
}
err := packer.Flush(ctx)
if err != nil {
t.Fatalf("failed to flush: %v", err)
}
blobs := packer.GetFinishedBlobs()
if len(blobs) != 1 {
t.Fatalf("expected 1 blob, got %d", len(blobs))
}
if len(blobs[0].Chunks) != testChunkCount {
t.Errorf("expected %d chunks in blob, got %d",
testChunkCount, len(blobs[0].Chunks))
}
// Verify offsets are correct
expectedOffset := int64(0)
for i, chunkRef := range blobs[0].Chunks {
if chunkRef.Offset != expectedOffset {
t.Errorf("chunk %d: expected offset %d, got %d",
i, expectedOffset, chunkRef.Offset)
}
if chunkRef.Length != testChunkSize {
t.Errorf("chunk %d: expected length %d, got %d",
i, testChunkSize, chunkRef.Length)
}
expectedOffset += chunkRef.Length
}
}
func TestPackerSizeLimit(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
const (
maxBlobSize = 5000 // 5KB max, forces multiple blobs
maxBlobawoOverhead = 6000 // allow some overhead over the limit
)
repos, packer := newTestPacker(t, maxBlobSize)
ctx := context.Background()
chunks := make([]*blob.ChunkRef, testChunkCount)
for i := range testChunkCount {
data := bytes.Repeat([]byte{byte(i)}, testChunkSize) // 1KB each
chunks[i] = makeChunk(t, repos, data)
}
blobCount := 0
// Add chunks and handle size limit errors
for _, chunk := range chunks {
err := packer.AddChunk(ctx, chunk)
if errors.Is(err, blob.ErrBlobSizeLimitExceeded) {
// Finalize current blob
err = packer.FinalizeBlob(ctx)
if err != nil {
t.Fatalf("failed to finalize blob: %v", err)
}
blobCount++
// Retry adding the chunk
err = packer.AddChunk(ctx, chunk)
if err != nil {
t.Fatalf("failed to add chunk after finalize: %v", err)
}
} else if err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
}
err := packer.Flush(ctx)
if err != nil {
t.Fatalf("failed to flush: %v", err)
}
blobs := packer.GetFinishedBlobs()
totalBlobs := blobCount + len(blobs)
if totalBlobs < 2 {
t.Errorf("expected multiple blobs due to size limit, got %d", totalBlobs)
}
// Verify each blob respects size limit (approximately)
for _, finished := range blobs {
if finished.Compressed > maxBlobawoOverhead {
t.Errorf("blob size %d exceeds limit", finished.Compressed)
}
}
}
func TestPackerEncryption(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
identity := parseTestIdentity(t)
repos, packer := newTestPacker(t, defaultMaxBlobSize)
ctx := context.Background()
data := bytes.Repeat([]byte("Test data for encryption!"), 100)
chunk := makeChunk(t, repos, data)
err := packer.AddChunk(ctx, chunk)
if err != nil {
t.Fatalf("failed to add chunk: %v", err)
}
err = packer.Flush(ctx)
if err != nil {
t.Fatalf("failed to flush: %v", err)
}
blobs := packer.GetFinishedBlobs()
if len(blobs) != 1 {
t.Fatalf("expected 1 blob, got %d", len(blobs))
}
decompressed := decryptAndDecompress(t, blobs[0].Data, identity)
if !bytes.Equal(decompressed, data) {
t.Error("decrypted and decompressed data doesn't match original")
}
}