persist metadata-only first batch when no symbols are selected

Author: j-rafique

supernode/adaptors/p2p.go

@@ -134,6 +134,32 @@ func (p *p2pImpl) storeCascadeSymbolsAndData(ctx context.Context, taskID, action
 	totalBytesStored := 0
 	metadataBytesStored := 0
 	firstBatchProcessed := false
+	if len(keys) == 0 && len(metadataFiles) > 0 {
+		logtrace.Info(ctx, "store: batch send (metadata-only)", logtrace.Fields{
+			"taskID":          taskID,
+			"metadata_count":  len(metadataFiles),
+			"metadata_bytes":  metadataBytes,
+			"metadata_mb_est": utils.BytesIntToMB(metadataBytes),
+		})
+
+		bctx, cancel := context.WithTimeout(ctx, storeBatchContextTimeout)
+		err = p.p2p.StoreBatch(bctx, metadataFiles, P2PDataRaptorQSymbol, taskID)
+		cancel()
+		if err != nil {
+			return totalSymbols, totalAvailable, fmt.Errorf("p2p store batch (metadata-only): %w", err)
+		}
+
+		logtrace.Info(ctx, "store: batch ok (metadata-only)", logtrace.Fields{
+			"taskID":         taskID,
+			"metadata_count": len(metadataFiles),
+			"metadata_bytes": metadataBytes,
+		})
+
+		totalBytesStored += metadataBytes
+		metadataBytesStored += metadataBytes
+		firstBatchProcessed = true
+	}
+
 	for start := 0; start < len(keys); {
 		end := min(start+loadSymbolsBatchSize, len(keys))
 		batch := keys[start:end]

supernode/adaptors/p2p_test.go

@@ -5,8 +5,10 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/LumeraProtocol/supernode/v2/p2p"
+	"github.com/LumeraProtocol/supernode/v2/pkg/codec"
 	"github.com/LumeraProtocol/supernode/v2/pkg/storage/rqstore"
 	"go.uber.org/mock/gomock"
 )
@@ -18,6 +20,35 @@ type clientWithPeersCount struct {
 
 func (c clientWithPeersCount) PeersCount() int { return c.peers }
 
+type testP2PClient struct {
+	storeBatchFn func(ctx context.Context, values [][]byte, typ int, taskID string) error
+}
+
+func (c testP2PClient) Retrieve(context.Context, string, ...bool) ([]byte, error) { return nil, nil }
+func (c testP2PClient) BatchRetrieve(context.Context, []string, int, string, ...bool) (map[string][]byte, error) {
+	return nil, nil
+}
+func (c testP2PClient) BatchRetrieveStream(context.Context, []string, int32, string, func(string, []byte) error, ...bool) (int32, error) {
+	return 0, nil
+}
+func (c testP2PClient) Store(context.Context, []byte, int) (string, error) { return "", nil }
+func (c testP2PClient) StoreBatch(ctx context.Context, values [][]byte, typ int, taskID string) error {
+	if c.storeBatchFn != nil {
+		return c.storeBatchFn(ctx, values, typ, taskID)
+	}
+	return nil
+}
+func (c testP2PClient) Delete(context.Context, string) error { return nil }
+func (c testP2PClient) Stats(context.Context) (*p2p.StatsSnapshot, error) { return nil, nil }
+func (c testP2PClient) NClosestNodes(context.Context, int, string, ...string) []string { return nil }
+func (c testP2PClient) NClosestNodesWithIncludingNodeList(context.Context, int, string, []string, []string) []string {
+	return nil
+}
+func (c testP2PClient) LocalStore(context.Context, string, []byte) (string, error) { return "", nil }
+func (c testP2PClient) DisableKey(context.Context, string) error { return nil }
+func (c testP2PClient) EnableKey(context.Context, string) error { return nil }
+func (c testP2PClient) GetLocalKeys(context.Context, *time.Time, time.Time) ([]string, error) { return nil, nil }
+
 func TestStoreArtefacts_ZeroPeers_ReturnsError(t *testing.T) {
 	svc := NewP2PService(clientWithPeersCount{peers: 0}, nil)
 
@@ -52,3 +83,74 @@ func TestStoreArtefacts_PeersPresent_DoesNotTripGuard(t *testing.T) {
 	}
 }
 
+func TestStoreCascadeSymbolsAndData_MetadataOnlyBatchWhenNoSymbols(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+
+	store := rqstore.NewMockStore(ctrl)
+	store.EXPECT().StoreSymbolDirectory("task", "").Return(nil)
+	store.EXPECT().UpdateIsFirstBatchStored("task").Return(nil)
+
+	metadata := [][]byte{[]byte("index-bytes"), []byte("layout-bytes")}
+	storeBatchCalls := 0
+	client := testP2PClient{storeBatchFn: func(_ context.Context, values [][]byte, typ int, taskID string) error {
+		storeBatchCalls++
+		if typ != P2PDataRaptorQSymbol {
+			t.Fatalf("unexpected type: %d", typ)
+		}
+		if taskID != "task" {
+			t.Fatalf("unexpected taskID: %s", taskID)
+		}
+		if len(values) != len(metadata) {
+			t.Fatalf("expected %d metadata values, got %d", len(metadata), len(values))
+		}
+		for i := range metadata {
+			if string(values[i]) != string(metadata[i]) {
+				t.Fatalf("metadata payload mismatch at %d", i)
+			}
+		}
+		return nil
+	}}
+
+	svc := &p2pImpl{p2p: client, rqStore: store}
+	stored, total, err := svc.storeCascadeSymbolsAndData(context.Background(), "task", "action", "", metadata, codec.Layout{Blocks: []codec.Block{{BlockID: 0}}})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if stored != 0 {
+		t.Fatalf("expected 0 stored symbols, got %d", stored)
+	}
+	if total != 0 {
+		t.Fatalf("expected 0 total symbols, got %d", total)
+	}
+	if storeBatchCalls != 1 {
+		t.Fatalf("expected StoreBatch to be called once, got %d", storeBatchCalls)
+	}
+}
+
+func TestStoreCascadeSymbolsAndData_MetadataOnlyBatchFailureSkipsFirstBatchFlag(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+
+	store := rqstore.NewMockStore(ctrl)
+	store.EXPECT().StoreSymbolDirectory("task", "").Return(nil)
+	store.EXPECT().UpdateIsFirstBatchStored("task").Times(0)
+
+	metadata := [][]byte{[]byte("index-bytes")}
+	client := testP2PClient{storeBatchFn: func(_ context.Context, values [][]byte, typ int, taskID string) error {
+		if typ != P2PDataRaptorQSymbol || taskID != "task" || len(values) != 1 {
+			t.Fatalf("unexpected StoreBatch args")
+		}
+		return errors.New("p2p down")
+	}}
+
+	svc := &p2pImpl{p2p: client, rqStore: store}
+	_, _, err := svc.storeCascadeSymbolsAndData(context.Background(), "task", "action", "", metadata, codec.Layout{Blocks: []codec.Block{{BlockID: 0}}})
+	if err == nil {
+		t.Fatalf("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "metadata-only") {
+		t.Fatalf("expected metadata-only path error, got: %v", err)
+	}
+}
+