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

Apply linter autofixes: internal/blob (refs #61)

Browse Source
This commit is contained in:
Jeffrey Paul
2026-08-07 16:53:19 +00:00
parent d34868a9c4
commit 5e4df7d04f
2 changed files with 81 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
internal/blob/packer.go
View File

@@ -18,6 +18,7 @@ import (
"context"
"database/sql"
"encoding/hex"
"errors"
"fmt"
"io"
"sync"
@@ -124,6 +125,7 @@ type BlobChunkRef struct {
// BlobWithReader wraps a FinishedBlob with its data reader
type BlobWithReader struct {
*FinishedBlob
Reader io.ReadSeeker
TempFile afero.File // Optional, only set for disk-based blobs
InsertedChunkHashes []string // Chunk hashes that were inserted to DB with this blob
@@ -134,14 +136,17 @@ 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, fmt.Errorf("recipients are required - blobs must be encrypted")
return nil, errors.New("recipients are required - blobs must be encrypted")
}
if cfg.MaxBlobSize <= 0 {
return nil, fmt.Errorf("max blob size must be positive")
return nil, errors.New("max blob size must be positive")
}
if cfg.Fs == nil {
return nil, fmt.Errorf("filesystem is required")
return nil, errors.New("filesystem is required")
}
return &Packer{
maxBlobSize: cfg.MaxBlobSize,
compressionLevel: cfg.CompressionLevel,
@@ -160,6 +165,7 @@ func NewPacker(cfg PackerConfig) (*Packer, error) {
func (p *Packer) SetBlobHandler(handler BlobHandler) {
p.mu.Lock()
defer p.mu.Unlock()
p.blobHandler = handler
}
@@ -169,6 +175,7 @@ func (p *Packer) SetBlobHandler(handler BlobHandler) {
func (p *Packer) AddPendingChunk(hash string, size int64) {
p.mu.Lock()
defer p.mu.Unlock()
p.pendingChunks = append(p.pendingChunks, PendingChunk{Hash: hash, Size: size})
}
@@ -183,7 +190,8 @@ func (p *Packer) AddChunk(chunk *ChunkRef) error {
// Initialize new blob if needed
if p.currentBlob == nil {
if err := p.startNewBlob(); err != nil {
err := p.startNewBlob()
if err != nil {
return fmt.Errorf("starting new blob: %w", err)
}
}
@@ -202,7 +210,8 @@ func (p *Packer) AddChunk(chunk *ChunkRef) error {
}
// Add chunk to current blob
if err := p.addChunkToCurrentBlob(chunk); err != nil {
err := p.addChunkToCurrentBlob(chunk)
if err != nil {
return err
}
@@ -218,7 +227,8 @@ func (p *Packer) Flush() error {
defer p.mu.Unlock()
if p.currentBlob != nil && len(p.currentBlob.chunks) > 0 {
if err := p.finalizeCurrentBlob(); err != nil {
err := p.finalizeCurrentBlob()
if err != nil {
return fmt.Errorf("finalizing blob: %w", err)
}
}
@@ -253,6 +263,7 @@ func (p *Packer) GetFinishedBlobs() []*FinishedBlob {
blobs := p.finishedBlobs
p.finishedBlobs = make([]*FinishedBlob, 0)
return blobs
}
@@ -267,6 +278,7 @@ func (p *Packer) startNewBlob() error {
if err != nil {
return fmt.Errorf("parsing blob ID: %w", err)
}
blob := &database.Blob{
ID: blobIDTyped,
Hash: types.BlobHash("temp-placeholder-" + blobID), // Temporary placeholder until finalized
@@ -276,6 +288,7 @@ func (p *Packer) startNewBlob() error {
CompressedSize: 0,
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 {
@@ -294,6 +307,7 @@ func (p *Packer) startNewBlob() error {
if err != nil {
_ = tempFile.Close()
_ = p.fs.Remove(tempFile.Name())
return fmt.Errorf("creating blobgen writer: %w", err)
}
@@ -308,6 +322,7 @@ func (p *Packer) startNewBlob() error {
}
log.Debug("Created new blob container", "blob_id", blobID, "temp_file", tempFile.Name())
return nil
}
@@ -316,6 +331,7 @@ 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)
return nil
}
@@ -403,24 +419,31 @@ func (p *Packer) finalizeCurrentBlob() error {
func (p *Packer) closeBlobWriter() (string, int64, error) {
if err := p.currentBlob.writer.Close(); err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("closing blobgen writer: %w", err)
}
if err := p.currentBlob.tempFile.Sync(); err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("syncing temp file: %w", err)
}
finalSize, err := p.currentBlob.tempFile.Seek(0, io.SeekCurrent)
if err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("getting file size: %w", err)
}
if _, err := p.currentBlob.tempFile.Seek(0, io.SeekStart); err != nil {
p.cleanupTempFile()
return "", 0, fmt.Errorf("seeking to start: %w", err)
}
finalHash := p.currentBlob.writer.Sum256()
return hex.EncodeToString(finalHash), finalSize, nil
}
@@ -432,6 +455,7 @@ func (p *Packer) buildChunkRefs() []*BlobChunkRef {
ChunkHash: chunk.Hash, Offset: chunk.Offset, Length: chunk.Size,
})
}
return refs
}
@@ -444,13 +468,16 @@ func (p *Packer) commitBlobToDatabase(blobHash string, finalSize int64, chunksTo
blobIDTyped, parseErr := types.ParseBlobID(p.currentBlob.id)
if parseErr != nil {
p.cleanupTempFile()
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}
if err := p.repos.Chunks.Create(ctx, tx, dbChunk); err != nil {
err := p.repos.Chunks.Create(ctx, tx, dbChunk)
if err != nil {
return fmt.Errorf("creating chunk: %w", err)
}
}
@@ -460,7 +487,9 @@ func (p *Packer) commitBlobToDatabase(blobHash string, finalSize int64, chunksTo
BlobID: blobIDTyped, ChunkHash: types.ChunkHash(chunk.Hash),
Offset: chunk.Offset, Length: chunk.Size,
}
if err := p.repos.BlobChunks.Create(ctx, tx, blobChunk); err != nil {
err := p.repos.BlobChunks.Create(ctx, tx, blobChunk)
if err != nil {
return fmt.Errorf("creating blob_chunk: %w", err)
}
}
@@ -469,11 +498,13 @@ func (p *Packer) commitBlobToDatabase(blobHash string, finalSize int64, chunksTo
})
if err != nil {
p.cleanupTempFile()
return fmt.Errorf("finalizing blob transaction: %w", err)
}
log.Debug("Committed blob transaction",
"chunks_inserted", len(chunksToInsert), "blob_chunks_inserted", len(p.currentBlob.chunks))
return nil
}
@@ -482,6 +513,7 @@ func (p *Packer) deliverFinishedBlob(finished *FinishedBlob, insertedChunkHashes
if p.blobHandler != nil {
if _, err := p.currentBlob.tempFile.Seek(0, io.SeekStart); err != nil {
p.cleanupTempFile()
return fmt.Errorf("seeking for handler: %w", err)
}
@@ -492,30 +524,39 @@ func (p *Packer) deliverFinishedBlob(finished *FinishedBlob, insertedChunkHashes
InsertedChunkHashes: insertedChunkHashes,
}
if err := p.blobHandler(blobWithReader); err != nil {
err := p.blobHandler(blobWithReader)
if err != nil {
p.cleanupTempFile()
return fmt.Errorf("blob handler failed: %w", err)
}
p.currentBlob = nil
return nil
}
// 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 {
p.cleanupTempFile()
return fmt.Errorf("seeking to read data: %w", err)
}
data, err := io.ReadAll(p.currentBlob.tempFile)
if err != nil {
p.cleanupTempFile()
return fmt.Errorf("reading blob data: %w", err)
}
finished.Data = data
p.finishedBlobs = append(p.finishedBlobs, finished)
p.cleanupTempFile()
p.currentBlob = nil
return nil
}
@@ -532,13 +573,15 @@ func (p *Packer) cleanupTempFile() {
func (p *Packer) PackChunks(chunks []*ChunkRef) error {
for _, chunk := range chunks {
err := p.AddChunk(chunk)
if err == ErrBlobSizeLimitExceeded {
if errors.Is(err, ErrBlobSizeLimitExceeded) {
// Finalize current blob and retry
if err := p.FinalizeBlob(); err != nil {
err := p.FinalizeBlob()
if err != nil {
return fmt.Errorf("finalizing blob before retry: %w", err)
}
// Retry the chunk
if err := p.AddChunk(chunk); err != nil {
err = p.AddChunk(chunk)
if err != nil {
return fmt.Errorf("adding chunk %s after finalize: %w", chunk.Hash, err)
}
} else if err != nil {

31
internal/blob/packer_test.go
View File

@@ -6,6 +6,7 @@ import (
"crypto/sha256"
"database/sql"
"encoding/hex"
"errors"
"io"
"testing"
@@ -40,6 +41,7 @@ func TestPacker(t *testing.T) {
t.Fatalf("failed to create test db: %v", err)
}
defer func() { _ = db.Close() }()
repos := database.NewRepositories(db)
cfg := PackerConfig{
@@ -49,6 +51,7 @@ func TestPacker(t *testing.T) {
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
@@ -64,6 +67,7 @@ func TestPacker(t *testing.T) {
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)
})
@@ -130,6 +134,7 @@ func TestPacker(t *testing.T) {
t.Fatalf("failed to create test db: %v", err)
}
defer func() { _ = db.Close() }()
repos := database.NewRepositories(db)
cfg := PackerConfig{
@@ -139,6 +144,7 @@ func TestPacker(t *testing.T) {
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
@@ -146,7 +152,8 @@ func TestPacker(t *testing.T) {
// Create multiple small chunks
chunks := make([]*ChunkRef, 10)
for i := 0; i < 10; i++ {
for i := range 10 {
data := bytes.Repeat([]byte{byte(i)}, 1000)
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
@@ -156,6 +163,7 @@ func TestPacker(t *testing.T) {
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)
})
@@ -198,9 +206,11 @@ func TestPacker(t *testing.T) {
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
}
})
@@ -212,6 +222,7 @@ func TestPacker(t *testing.T) {
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
@@ -222,6 +233,7 @@ func TestPacker(t *testing.T) {
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
@@ -229,7 +241,8 @@ func TestPacker(t *testing.T) {
// Create chunks that will exceed the limit
chunks := make([]*ChunkRef, 10)
for i := 0; i < 10; i++ {
for i := range 10 {
data := bytes.Repeat([]byte{byte(i)}, 1000) // 1KB each
hash := sha256.Sum256(data)
hashStr := hex.EncodeToString(hash[:])
@@ -239,6 +252,7 @@ func TestPacker(t *testing.T) {
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)
})
@@ -257,14 +271,17 @@ func TestPacker(t *testing.T) {
// Add chunks and handle size limit errors
for _, chunk := range chunks {
err := packer.AddChunk(chunk)
if err == ErrBlobSizeLimitExceeded {
if errors.Is(err, ErrBlobSizeLimitExceeded) {
// Finalize current blob
if err := packer.FinalizeBlob(); err != nil {
err := packer.FinalizeBlob()
if err != nil {
t.Fatalf("failed to finalize blob: %v", err)
}
blobCount++
// Retry adding the chunk
if err := packer.AddChunk(chunk); err != nil {
err = packer.AddChunk(chunk)
if err != nil {
t.Fatalf("failed to add chunk after finalize: %v", err)
}
} else if err != nil {
@@ -301,6 +318,7 @@ func TestPacker(t *testing.T) {
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)
@@ -311,6 +329,7 @@ func TestPacker(t *testing.T) {
Repositories: repos,
Fs: afero.NewMemMapFs(),
}
packer, err := NewPacker(cfg)
if err != nil {
t.Fatalf("failed to create packer: %v", err)
@@ -326,6 +345,7 @@ func TestPacker(t *testing.T) {
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)
})
@@ -342,6 +362,7 @@ func TestPacker(t *testing.T) {
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)
}