Codeq settings

Author: mateeullahmalik

pkg/codec/decode.go

@@ -39,11 +39,28 @@ func (rq *raptorQ) Decode(ctx context.Context, req DecodeRequest) (DecodeRespons
 	defer processor.Free()
 
 	symbolsDir := filepath.Join(rq.symbolsBaseDir, req.ActionID)
+	// Ensure a clean scratch directory (avoid contamination from previous attempts)
+	if err := os.RemoveAll(symbolsDir); err != nil {
+		fields[logtrace.FieldError] = err.Error()
+		return DecodeResponse{}, fmt.Errorf("cleanup decode dir %s: %w", symbolsDir, err)
+	}
 	if err := os.MkdirAll(symbolsDir, 0o755); err != nil {
 		fields[logtrace.FieldError] = err.Error()
 		return DecodeResponse{}, fmt.Errorf("mkdir %s: %w", symbolsDir, err)
 	}
 
+	// Validate layout before writing any symbols
+	if len(req.Layout.Blocks) == 0 {
+		fields[logtrace.FieldError] = "empty layout"
+		return DecodeResponse{}, fmt.Errorf("invalid layout: no blocks present")
+	}
+	for _, blk := range req.Layout.Blocks {
+		if len(blk.Symbols) == 0 {
+			fields[logtrace.FieldError] = fmt.Sprintf("block_%d has no symbols", blk.BlockID)
+			return DecodeResponse{}, fmt.Errorf("invalid layout: block %d has no symbols", blk.BlockID)
+		}
+	}
+
 	// Build symbol->block mapping from layout and ensure block directories exist
 	symbolToBlock := make(map[string]int)
 	for _, blk := range req.Layout.Blocks {

pkg/codec/raptorq.go

@@ -17,7 +17,7 @@ const (
 	// Limit RaptorQ processor memory usage to ~2 GiB
 	rqMaxMemoryMB uint64 = 2 * 1024 // MB
 	// Concurrency tuned for 2 GiB limit and typical 8+ core CPUs
-	rqConcurrency uint64 = 6
+	rqConcurrency uint64 = 1
 	// Target single-block output for up to 1 GiB files with padding headroom (~1.25 GiB)
 	rqBlockSize int = 1280 * 1024 * 1024 // bytes (1,280 MiB)
 )

supernode/services/cascade/download.go

@@ -163,7 +163,7 @@ func (task *CascadeRegistrationTask) restoreFileFromLayout(
 	if targetRequiredCount < 1 && totalSymbols > 0 {
 		targetRequiredCount = 1
 	}
-	logtrace.Info(ctx, "Retrieving all symbols for decode", fields)
+	logtrace.Info(ctx, "Retrieving target-required symbols for decode", fields)
 
 	// Enable retrieve metrics capture for this action
 	cm.StartRetrieveCapture(actionID)
@@ -173,7 +173,13 @@ func (task *CascadeRegistrationTask) restoreFileFromLayout(
 	retrieveStart := time.Now()
 	// Tag context with metrics task ID (actionID)
 	ctxRetrieve := cm.WithTaskID(ctx, actionID)
-	symbols, err := task.P2PClient.BatchRetrieve(ctxRetrieve, allSymbols, totalSymbols, actionID)
+	// Retrieve only a fraction of symbols (targetRequiredCount) based on redundancy
+	// The DHT will short-circuit once it finds the required number across the provided keys
+	reqCount := targetRequiredCount
+	if reqCount > totalSymbols {
+		reqCount = totalSymbols
+	}
+	symbols, err := task.P2PClient.BatchRetrieve(ctxRetrieve, allSymbols, reqCount, actionID)
 	if err != nil {
 		fields[logtrace.FieldError] = err.Error()
 		logtrace.Error(ctx, "batch retrieve failed", fields)