Merge pull request #3 from hcp-uw/shardLib

First steps into file sharding

Author: Gjones747

.github/workflows/go-tests.yaml

@@ -0,0 +1,61 @@
+name: Go CI
+
+on:
+  push:
+    branches: [ main, master ]
+  pull_request:
+    branches: [ main, master ]
+
+jobs:
+  ci:
+    runs-on: ubuntu-latest
+
+    steps:
+      # --- Checkout code (Pinned to v4) ---
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      # --- Setup Go (Pinned to v6) ---
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: v1.25.1
+
+      # --- Cache Go modules (Pinned to v4) ---
+      - name: Cache Go modules
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/go/pkg/mod
+            ~/.cache/go-build
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ runner.os }}-go-
+
+      # --- Download Dependencies (Recommended Practice) ---
+      - name: Download Go modules
+        run: go mod download
+
+          
+      # --- Run tests and collect coverage ---
+      - name: Run tests with coverage
+        run: |
+          go test -v -coverprofile=coverage.out ./...
+          go tool cover -func=coverage.out
+
+      # --- Generate coverage badge (Pinned to v2) ---
+      - name: Generate coverage badge
+        uses: tj-actions/coverage-badge-go@v2
+        if: github.ref == 'refs/heads/main'
+        with:
+          filename: coverage.out
+
+      # --- Commit coverage badge (main only) ---
+      - name: Commit coverage badge
+        if: github.ref == 'refs/heads/main'
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          git add coverage-badge.svg
+          git commit -m "chore: update coverage badge" || echo "No changes to commit"
+          git push

.gitignore

@@ -0,0 +1,3 @@
+notes.md
+files
+output

cmd/mosaic-node/main.go

@@ -1,7 +1,39 @@
 package main
 
-import "fmt"
+import (
+	"log"
+	"os"
+
+	"github.com/hcp-uw/mosaic/internal/encoding"
+)
 
 func main() {
-	fmt.Println("welcome to mosaic")
+	//for testing purposes rn
+
+	// fileSize := 1267513984
+	file, err := os.ReadFile("output_file.jpg")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	fileSize := len(file)
+
+	encoder, err := encoding.NewEncoder(8, 4, "./files", "./files/.bin")
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	//err = encoder.EncodeFile("/pictures/pic.jpg")
+	//if err != nil {
+	//	fmt.Println(err)
+	//}
+
+	//fmt.Println(fileSize)
+
+	err = encoder.DecodeShards("pictures/pic.jpg", fileSize)
+	if err != nil {
+		log.Fatal(err)
+	}
+
 }

go.mod

@@ -1,3 +1,9 @@
 module github.com/hcp-uw/mosaic
 
 go 1.25.1
+
+require (
+	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
+	github.com/klauspost/reedsolomon v1.12.5 // indirect
+	golang.org/x/sys v0.37.0 // indirect
+)

go.sum

@@ -0,0 +1,6 @@
+github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
+github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
+github.com/klauspost/reedsolomon v1.12.5 h1:4cJuyH926If33BeDgiZpI5OU0pE+wUHZvMSyNGqN73Y=
+github.com/klauspost/reedsolomon v1.12.5/go.mod h1:LkXRjLYGM8K/iQfujYnaPeDmhZLqkrGUyG9p7zs5L68=
+golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
+golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=

internal/encoding/decode.go

@@ -0,0 +1,165 @@
+package encoding
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+)
+
+// has to remember file length
+// relativePath is the path inside the IN folder where incoming shards are stored
+// should be a directory not a file
+func (e *Encoder) DecodeShards(relativePath string, fileLength int) error {
+	totalShards := e.parity + e.shards
+	fileName := filepath.Base(relativePath)
+	fileNameNoExt := strings.TrimSuffix(fileName, filepath.Ext(fileName))
+
+	type shardResult struct {
+		index  int
+		result []byte
+	}
+
+	shardFiles := make([]*os.File, totalShards)
+	for i := 0; i < totalShards; i++ {
+		shardName := fmt.Sprintf("shard%d_%s.dat", i, fileNameNoExt)
+		shardPath := filepath.Join(e.dirIn, relativePath, shardName)
+
+		file, err := os.Open(shardPath)
+		if err != nil {
+			shardFiles[i] = nil
+			continue
+		}
+		shardFiles[i] = file
+	}
+
+	// makes sure outpath exists if not it creates it
+	outPath := filepath.Dir(filepath.Join(e.dirOut, relativePath))
+	err := os.MkdirAll(outPath, 0755)
+	if err != nil {
+		log.Fatal("here")
+		return err
+	}
+
+	// on first run through it finds the file extension creates the file then further
+	// writes to that file
+	var outFile *os.File
+	writtenBytes := 0
+	fileCreated := false
+	for {
+		if fileLength <= writtenBytes {
+			break
+		}
+
+		shardResults := make(chan shardResult, totalShards)
+		shardArray := make([][]byte, totalShards)
+		var shardReaders sync.WaitGroup
+
+		for index, file := range shardFiles {
+			if file == nil {
+				continue
+			}
+
+			shardReaders.Add(1)
+			go func(index int, file *os.File) {
+				defer shardReaders.Done()
+				shard := make([]byte, e.blockSize)
+
+				io.ReadFull(file, shard)
+				// ts silent error is not that tuff
+
+				shardResults <- shardResult{index: index, result: shard}
+
+			}(index, file)
+		}
+
+		go func() {
+			shardReaders.Wait()
+			close(shardResults)
+		}()
+
+		var dataReadCount int
+		for res := range shardResults {
+			shardArray[res.index] = res.result
+			if res.result != nil {
+				dataReadCount++
+			}
+		}
+
+		// CRITICAL FIX: Exit condition 2: If zero non-nil blocks were read in this iteration,
+		// it means all active shards are exhausted. Break the decoding loop to prevent
+		// continuous reconstruction failure on empty data.
+		if dataReadCount == 0 {
+			break
+		}
+
+		err := e.encoder.ReconstructData(shardArray)
+		if err != nil {
+			return err
+		}
+
+		var buf bytes.Buffer
+
+		// works except for padding
+		err = e.encoder.Join(&buf, shardArray, e.shards*e.blockSize)
+		joinedBytes := buf.Bytes()
+		if !fileCreated {
+
+			fileExtension := detectExtension(joinedBytes)
+			fileOutDir := filepath.Join(e.dirOut, filepath.Dir(relativePath))
+			fileOutPath := filepath.Join(fileOutDir, fileNameNoExt+fileExtension)
+			os.MkdirAll(fileOutDir, 0755)
+
+			outFile, err = os.Create(fileOutPath)
+			if err != nil {
+				return err
+			}
+			defer outFile.Close()
+
+			fileCreated = true
+		}
+
+		remaining := fileLength - writtenBytes
+		toWrite := len(joinedBytes)
+		if remaining < toWrite {
+			toWrite = remaining
+		}
+
+		totalWritten := 0
+		for totalWritten < toWrite {
+			n, err := outFile.Write(joinedBytes[totalWritten:toWrite])
+			if err != nil {
+				return err
+			}
+			totalWritten += n
+			writtenBytes += n
+		}
+	}
+
+	return nil
+}
+
+func detectExtension(data []byte) string {
+	if len(data) > 512 {
+		data = data[:512] // only need first 512 bytes
+	}
+	mimeType := http.DetectContentType(data)
+
+	switch mimeType {
+	case "image/png":
+		return ".png"
+	case "image/jpeg":
+		return ".jpg"
+	case "application/pdf":
+		return ".pdf"
+	case "text/plain; charset=utf-8":
+		return ".txt"
+	default:
+		return ".bin" // fallback
+	}
+}

internal/encoding/decode_test.go

@@ -0,0 +1,95 @@
+ package encoding
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// TestDecodeShards_Performance runs full encode→decode and prints throughput (MB/s)
+func TestDecodeShards_Performance(t *testing.T) {
+	tmpIn := t.TempDir()
+	tmpOut := t.TempDir()
+
+	// prepare a moderately large file (e.g., 50 MB)
+	fileSize := 50 * 1024 * 1024
+	fileName := "bigfile.txt"
+	inFilePath := filepath.Join(tmpOut, fileName)
+	data := make([]byte, fileSize)
+	for i := range data {
+		data[i] = byte(i % 256)
+	}
+
+	if err := os.WriteFile(inFilePath, data, 0644); err != nil {
+		t.Fatalf("failed to write input file: %v", err)
+	}
+
+	// create encoder
+	enc, err := NewEncoder(4, 2, tmpOut, tmpIn)
+	if err != nil {
+		t.Fatalf("failed to create encoder: %v", err)
+	}
+
+	// make .bin dir for encoded output
+	if err := os.MkdirAll(filepath.Join(tmpOut, ".bin"), 0755); err != nil {
+		t.Fatalf("failed to make .bin dir: %v", err)
+	}
+
+	// encode
+	startEncode := time.Now()
+	if err := enc.EncodeFile(fileName); err != nil {
+		t.Fatalf("EncodeFile failed: %v", err)
+	}
+	encodeElapsed := time.Since(startEncode).Seconds()
+
+	// move shards to decoder input (.bin/<fileName>)
+	shardDir := filepath.Join(tmpOut, ".bin", fileName)
+	shards, err := os.ReadDir(shardDir)
+	if err != nil {
+		t.Fatalf("failed to read shard dir: %v", err)
+	}
+	targetShardDir := filepath.Join(tmpIn, fileName)
+	if err := os.MkdirAll(targetShardDir, 0755); err != nil {
+		t.Fatalf("failed to create shard dir: %v", err)
+	}
+	for _, shard := range shards {
+		src := filepath.Join(shardDir, shard.Name())
+		dst := filepath.Join(targetShardDir, shard.Name())
+		b, err := os.ReadFile(src)
+		if err != nil {
+			t.Fatalf("failed to read shard: %v", err)
+		}
+		if err := os.WriteFile(dst, b, 0644); err != nil {
+			t.Fatalf("failed to write shard: %v", err)
+		}
+	}
+
+	// decode
+	startDecode := time.Now()
+	if err := enc.DecodeShards(fileName, len(data)); err != nil {
+		t.Fatalf("DecodeShards failed: %v", err)
+	}
+	decodeElapsed := time.Since(startDecode).Seconds()
+
+	// verify output
+	decodedPath := filepath.Join(tmpOut, fileName)
+	decodedData, err := os.ReadFile(decodedPath)
+	if err != nil {
+		t.Fatalf("failed to read decoded file: %v", err)
+	}
+
+	if !bytes.Equal(decodedData, data) {
+		t.Fatalf("decoded file does not match original data")
+	}
+
+	// calculate and log throughput
+	mb := float64(fileSize) / (1024 * 1024)
+	encodeThroughput := mb / encodeElapsed
+	decodeThroughput := mb / decodeElapsed
+
+	t.Logf("Encode: %.2f MB in %.2f s (%.2f MB/s)", mb, encodeElapsed, encodeThroughput)
+	t.Logf("Decode: %.2f MB in %.2f s (%.2f MB/s)", mb, decodeElapsed, decodeThroughput)
+}
+