remove split blobs

Author: julienrbrt

block/internal/da/fiber_client.go

@@ -2,7 +2,6 @@ package da
 
 import (
 	"context"
-	"encoding/binary"
 	"errors"
 	"fmt"
 	"sync"
@@ -253,21 +252,9 @@ loop:
 				},
 			}
 		}
-		split, splitErr := splitBlobs(blobData)
-		if splitErr != nil {
-			return datypes.ResultRetrieve{
-				BaseResult: datypes.BaseResult{
-					Code:      datypes.StatusError,
-					Message:   fmt.Sprintf("fiber decode failed for blob %x: %v", blobID, splitErr),
-					Height:    height,
-					Timestamp: time.Now(),
-				},
-			}
-		}
-		for _, b := range split {
-			ids = append(ids, blobID)
-			data = append(data, b)
-		}
+
+		ids = append(ids, blobID)
+		data = append(data, blobData)
 	}
 
 	return datypes.ResultRetrieve{
@@ -294,11 +281,8 @@ func (c *fiberDAClient) Get(ctx context.Context, ids []datypes.ID, _ []byte) ([]
 		if err != nil {
 			return nil, fmt.Errorf("fiber download failed for blob %x: %w", id, err)
 		}
-		split, splitErr := splitBlobs(data)
-		if splitErr != nil {
-			return nil, fmt.Errorf("fiber decode failed for blob %x: %w", id, splitErr)
-		}
-		res = append(res, split...)
+
+		res = append(res, data)
 	}
 
 	return res, nil
@@ -312,10 +296,6 @@ func (c *fiberDAClient) Subscribe(ctx context.Context, namespace []byte, _ bool)
 	go func() {
 		defer close(out)
 
-		// The outer DA Subscribe entry point does not expose a starting
-		// height, so start from the live tip (fromHeight=0). A future
-		// refactor that plumbs resume-from-height through datypes.DA can
-		// thread the value here.
 		blobCh, err := c.fiber.Listen(ctx, namespace[len(namespace)-10:], c.lastKnownDAHeight)
 		if err != nil {
 			c.logger.Error().Err(err).Msg("fiber listen failed")
@@ -337,17 +317,11 @@ func (c *fiberDAClient) Subscribe(ctx context.Context, namespace []byte, _ bool)
 					continue
 				}
 
-				split, splitErr := splitBlobs(blobData)
-				if splitErr != nil {
-					c.logger.Error().Err(splitErr).Bytes("blob_id", event.BlobID).Msg("failed to decode blob")
-					continue
-				}
-
 				select {
 				case out <- datypes.SubscriptionEvent{
 					Height:    event.Height,
 					Timestamp: time.Now(),
-					Blobs:     split,
+					Blobs:     [][]byte{blobData},
 				}:
 				case <-ctx.Done():
 					return
@@ -398,48 +372,6 @@ func (c *fiberDAClient) Validate(_ context.Context, ids []datypes.ID, proofs []d
 func (c *fiberDAClient) GetHeaderNamespace() []byte { return c.namespaceBz }
 func (c *fiberDAClient) GetDataNamespace() []byte   { return c.dataNamespaceBz }
 
-// splitBlobs decodes the legacy "count + per-item length" framing that
-// the previous Submit path used to pack multiple blobs into a single
-// Upload. The per-item-concurrent Submit path no longer writes that
-// framing — each item is uploaded raw — so any blob written by this
-// branch's Submit will fail to decode here.
-//
-// Callers (Retrieve / RetrieveBlobs / Get / Subscribe) therefore only
-// work for blobs written by the OLD code path, OR for the multi-item
-// header batches that still use it. Pair the format change with a
-// matching update to the read path before any node on this branch
-// tries to sync from another node on this branch.
-//
-// Tracked alongside the wire-format TODO on Submit (above).
-func splitBlobs(data []byte) ([][]byte, error) {
-	if len(data) == 0 {
-		return nil, nil
-	}
-	if len(data) < 4 {
-		return nil, fmt.Errorf("invalid blob encoding: header too short")
-	}
-
-	count := int(binary.BigEndian.Uint32(data))
-	off := 4
-	blobs := make([][]byte, 0, count)
-	for i := range count {
-		if off+4 > len(data) {
-			return nil, fmt.Errorf("invalid blob encoding: truncated length at index %d", i)
-		}
-		size := int(binary.BigEndian.Uint32(data[off:]))
-		off += 4
-		end := off + size
-		if end < off || end > len(data) {
-			return nil, fmt.Errorf("invalid blob encoding: truncated data at index %d", i)
-		}
-		blob := make([]byte, size)
-		copy(blob, data[off:end])
-		off = end
-		blobs = append(blobs, blob)
-	}
-	return blobs, nil
-}
-
 // Force Inclusion is disabled for Fiber PoC.
 func (c *fiberDAClient) HasForcedInclusionNamespace() bool   { return false }
 func (c *fiberDAClient) GetForcedInclusionNamespace() []byte { return nil }

block/internal/da/fiber_client_test.go

@@ -479,27 +479,3 @@ func (f *failOnNthUpload) Upload(ctx context.Context, namespace, data []byte) (f
 	}
 	return f.FiberClient.Upload(ctx, namespace, data)
 }
-
-func TestSplitBlobs_Empty(t *testing.T) {
-	got, err := splitBlobs(nil)
-	require.NoError(t, err)
-	require.Nil(t, got)
-
-	got, err = splitBlobs([]byte{})
-	require.NoError(t, err)
-	require.Nil(t, got)
-}
-
-func TestSplitBlobs_Truncated(t *testing.T) {
-	_, err := splitBlobs([]byte{0x01})
-	require.Error(t, err)
-
-	_, err = splitBlobs([]byte{0x00, 0x00, 0x00, 0x01, 0x00, 0x00})
-	require.Error(t, err)
-}
-
-func TestSplitBlobs_CountMismatch(t *testing.T) {
-	data := []byte{0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x01, 0x41}
-	_, err := splitBlobs(data)
-	require.Error(t, err)
-}