diff --git a/internal/blob/packer.go b/internal/blob/packer.go index 116c365..1e02096 100644 --- a/internal/blob/packer.go +++ b/internal/blob/packer.go @@ -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 { diff --git a/internal/blob/packer_test.go b/internal/blob/packer_test.go index dba194b..d44bc22 100644 --- a/internal/blob/packer_test.go +++ b/internal/blob/packer_test.go @@ -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) }