implement cascade action download functionality in action-sdk

Author: j-rafique

gen/supernode/action/cascade/service.pb.go

@@ -61,8 +61,6 @@ github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3
 github.com/DataDog/zstd v1.5.5 h1:oWf5W7GtOLgp6bciQYDmhHHjdhYkALu6S/5Ni9ZgSvQ=
 github.com/DataDog/zstd v1.5.5/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
-github.com/LumeraProtocol/lumera v1.5.0 h1:LDPtd155PjG/LKk34x/3vhC9H+J9tHoxwrcwRMG6jzM=
-github.com/LumeraProtocol/lumera v1.5.0/go.mod h1:c1M+sjewuCvxw+pznwlspUzenDJI8Y+suKB3RFKS2Wo=
 github.com/LumeraProtocol/rq-go v0.2.1 h1:8B3UzRChLsGMmvZ+UVbJsJj6JZzL9P9iYxbdUwGsQI4=
 github.com/LumeraProtocol/rq-go v0.2.1/go.mod h1:APnKCZRh1Es2Vtrd2w4kCLgAyaL5Bqrkz/BURoRJ+O8=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=

go.sum

@@ -2,6 +2,7 @@ package codec
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -17,8 +18,8 @@ type DecodeRequest struct {
 }
 
 type DecodeResponse struct {
-	Path       string
-	LayoutPath string
+	Path         string
+	DecodeTmpDir string
 }
 
 func (rq *raptorQ) Decode(ctx context.Context, req DecodeRequest) (DecodeResponse, error) {
@@ -52,14 +53,27 @@ func (rq *raptorQ) Decode(ctx context.Context, req DecodeRequest) (DecodeRespons
 	}
 	logtrace.Info(ctx, "symbols written to disk", fields)
 
+	// ---------- write layout.json ----------  ←★
+	layoutPath := filepath.Join(symbolsDir, "layout.json")
+	layoutBytes, err := json.Marshal(req.Layout)
+	if err != nil {
+		fields[logtrace.FieldError] = err.Error()
+		return DecodeResponse{}, fmt.Errorf("marshal layout: %w", err)
+	}
+	if err := os.WriteFile(layoutPath, layoutBytes, 0o644); err != nil {
+		fields[logtrace.FieldError] = err.Error()
+		return DecodeResponse{}, fmt.Errorf("write layout file: %w", err)
+	}
+	logtrace.Info(ctx, "layout.json written", fields)
+
 	// Decode
 	outputPath := filepath.Join(symbolsDir, "output")
-	if err := processor.DecodeSymbols(symbolsDir, outputPath, ""); err != nil {
+	if err := processor.DecodeSymbols(symbolsDir, outputPath, layoutPath); err != nil {
 		fields[logtrace.FieldError] = err.Error()
 		_ = os.Remove(outputPath)
 		return DecodeResponse{}, fmt.Errorf("raptorq decode: %w", err)
 	}
 
 	logtrace.Info(ctx, "RaptorQ decoding completed successfully", fields)
-	return DecodeResponse{Path: outputPath, LayoutPath: ""}, nil
+	return DecodeResponse{Path: outputPath, DecodeTmpDir: symbolsDir}, nil
 }

pkg/codec/decode.go

@@ -59,7 +59,7 @@ enum SupernodeEventType {
   RQID_VERIFIED = 9;
   ARTEFACTS_STORED = 10;
   ACTION_FINALIZED = 11;
-  Artefacts_Downloaded = 12;
+  ARTEFACTS_DOWNLOADED = 12;
 }
 
 message HealthCheckRequest {}

proto/supernode/action/cascade/service.proto

@@ -25,6 +25,7 @@ type Client interface {
 	GetTask(ctx context.Context, taskID string) (*task.TaskEntry, bool)
 	SubscribeToEvents(ctx context.Context, eventType event.EventType, handler event.Handler) error
 	SubscribeToAllEvents(ctx context.Context, handler event.Handler) error
+	DownloadCascade(ctx context.Context, actionID, outputPath string) (string, error)
 }
 
 // ClientImpl implements the Client interface
@@ -128,3 +129,25 @@ func (c *ClientImpl) SubscribeToAllEvents(ctx context.Context, handler event.Han
 
 	return nil
 }
+
+func (c *ClientImpl) DownloadCascade(
+	ctx context.Context,
+	actionID, outputPath string,
+) (string, error) {
+
+	if actionID == "" {
+		return "", fmt.Errorf("actionID is empty")
+	}
+
+	taskID, err := c.taskManager.CreateDownloadTask(ctx, actionID, outputPath)
+	if err != nil {
+		return "", fmt.Errorf("create download task: %w", err)
+	}
+
+	c.logger.Info(ctx, "cascade download task created",
+		"task_id", taskID,
+		"action_id", actionID,
+	)
+
+	return taskID, nil
+}

sdk/action/client.go

@@ -164,6 +164,98 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 	}, nil
 }
 
+// CascadeSupernodeDownload downloads a file from a supernode gRPC stream
+func (a *cascadeAdapter) CascadeSupernodeDownload(
+	ctx context.Context,
+	in *CascadeSupernodeDownloadRequest,
+	opts ...grpc.CallOption,
+) (*CascadeSupernodeDownloadResponse, error) {
+
+	ctx = net.AddCorrelationID(ctx)
+
+	// 1. Open gRPC stream (server-stream)
+	stream, err := a.client.Download(ctx, &cascade.DownloadRequest{
+		ActionId: in.ActionID,
+	}, opts...)
+	if err != nil {
+		a.logger.Error(ctx, "failed to create download stream",
+			"action_id", in.ActionID, "error", err)
+		return nil, err
+	}
+
+	// 2. Prepare destination file
+	outFile, err := os.Create(in.OutputPath)
+	if err != nil {
+		a.logger.Error(ctx, "failed to create output file",
+			"path", in.OutputPath, "error", err)
+		return nil, fmt.Errorf("create output file: %w", err)
+	}
+	defer outFile.Close()
+
+	var (
+		bytesWritten int64
+		chunkIndex   int
+	)
+
+	// 3. Receive streamed responses
+	for {
+		resp, err := stream.Recv()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("stream recv: %w", err)
+		}
+
+		switch x := resp.ResponseType.(type) {
+
+		// 3a. Progress / event message
+		case *cascade.DownloadResponse_Event:
+			a.logger.Info(ctx, "supernode event",
+				"event_type", x.Event.EventType,
+				"message", x.Event.Message,
+				"action_id", in.ActionID)
+
+			if in.EventLogger != nil {
+				in.EventLogger(ctx, toSdkEvent(x.Event.EventType), x.Event.Message, event.EventData{
+					event.KeyActionID:  in.ActionID,
+					event.KeyEventType: x.Event.EventType,
+					event.KeyMessage:   x.Event.Message,
+				})
+			}
+
+		// 3b. Actual data chunk
+		case *cascade.DownloadResponse_Chunk:
+			data := x.Chunk.Data
+			if len(data) == 0 {
+				continue
+			}
+			if _, err := outFile.Write(data); err != nil {
+				return nil, fmt.Errorf("write chunk: %w", err)
+			}
+
+			bytesWritten += int64(len(data))
+			chunkIndex++
+
+			a.logger.Debug(ctx, "received chunk",
+				"chunk_index", chunkIndex,
+				"chunk_size", len(data),
+				"bytes_written", bytesWritten)
+		}
+	}
+
+	a.logger.Info(ctx, "download complete",
+		"bytes_written", bytesWritten,
+		"path", in.OutputPath,
+		"action_id", in.ActionID)
+
+	return &CascadeSupernodeDownloadResponse{
+		Success:    true,
+		Message:    "artefact downloaded",
+		OutputPath: in.OutputPath,
+	}, nil
+}
+
 // toSdkEvent converts a supernode-side enum value into an internal SDK EventType.
 func toSdkEvent(e cascade.SupernodeEventType) event.EventType {
 	switch e {
@@ -189,6 +281,8 @@ func toSdkEvent(e cascade.SupernodeEventType) event.EventType {
 		return event.SupernodeArtefactsStored
 	case cascade.SupernodeEventType_ACTION_FINALIZED:
 		return event.SupernodeActionFinalized
+	case cascade.SupernodeEventType_ARTEFACTS_DOWNLOADED:
+		return event.SupernodeArtefactsDownloaded
 	default:
 		return event.SupernodeUnknown
 	}

sdk/adapters/supernodeservice/adapter.go

@@ -28,7 +28,21 @@ type CascadeSupernodeRegisterResponse struct {
 	TxHash  string
 }
 
+type CascadeSupernodeDownloadRequest struct {
+	ActionID    string
+	TaskID      string
+	OutputPath  string
+	EventLogger LoggerFunc
+}
+
+type CascadeSupernodeDownloadResponse struct {
+	Success    bool
+	Message    string
+	OutputPath string
+}
+
 //go:generate mockery --name=CascadeServiceClient --output=testutil/mocks --outpkg=mocks --filename=cascade_service_mock.go
 type CascadeServiceClient interface {
 	CascadeSupernodeRegister(ctx context.Context, in *CascadeSupernodeRegisterRequest, opts ...grpc.CallOption) (*CascadeSupernodeRegisterResponse, error)
+	CascadeSupernodeDownload(ctx context.Context, in *CascadeSupernodeDownloadRequest, opts ...grpc.CallOption) (*CascadeSupernodeDownloadResponse, error)
 }

sdk/adapters/supernodeservice/types.go

@@ -14,6 +14,7 @@ const (
 	KeyMessage          EventDataKey = "message"
 	KeyProgress         EventDataKey = "progress"
 	KeyEventType        EventDataKey = "event_type"
+	KeyOutputPath       EventDataKey = "output_path"
 
 	// Task specific keys
 	KeyTaskID   EventDataKey = "task_id"

sdk/event/keys.go

@@ -23,21 +23,27 @@ const (
 	SDKTaskTxHashReceived     EventType = "sdk:txhash_received"
 	SDKTaskCompleted          EventType = "sdk:completed"
 	SDKTaskFailed             EventType = "sdk:failed"
+
+	SDKDownloadAttempt    EventType = "sdk:download_attempt"
+	SDKDownloadFailure    EventType = "sdk:download_failure"
+	SDKOutputPathReceived EventType = "sdk:output_path_received"
+	SDKDownloadSuccessful EventType = "sdk:download_successful"
 )
 
 const (
-	SupernodeActionRetrieved   EventType = "supernode:action_retrieved"
-	SupernodeActionFeeVerified EventType = "supernode:action_fee_verified"
-	SupernodeTopCheckPassed    EventType = "supernode:top_check_passed"
-	SupernodeMetadataDecoded   EventType = "supernode:metadata_decoded"
-	SupernodeDataHashVerified  EventType = "supernode:data_hash_verified"
-	SupernodeInputEncoded      EventType = "supernode:input_encoded"
-	SupernodeSignatureVerified EventType = "supernode:signature_verified"
-	SupernodeRQIDGenerated     EventType = "supernode:rqid_generated"
-	SupernodeRQIDVerified      EventType = "supernode:rqid_verified"
-	SupernodeArtefactsStored   EventType = "supernode:artefacts_stored"
-	SupernodeActionFinalized   EventType = "supernode:action_finalized"
-	SupernodeUnknown           EventType = "supernode:unknown"
+	SupernodeActionRetrieved     EventType = "supernode:action_retrieved"
+	SupernodeActionFeeVerified   EventType = "supernode:action_fee_verified"
+	SupernodeTopCheckPassed      EventType = "supernode:top_check_passed"
+	SupernodeMetadataDecoded     EventType = "supernode:metadata_decoded"
+	SupernodeDataHashVerified    EventType = "supernode:data_hash_verified"
+	SupernodeInputEncoded        EventType = "supernode:input_encoded"
+	SupernodeSignatureVerified   EventType = "supernode:signature_verified"
+	SupernodeRQIDGenerated       EventType = "supernode:rqid_generated"
+	SupernodeRQIDVerified        EventType = "supernode:rqid_verified"
+	SupernodeArtefactsStored     EventType = "supernode:artefacts_stored"
+	SupernodeActionFinalized     EventType = "supernode:action_finalized"
+	SupernodeArtefactsDownloaded EventType = "supernode:artefacts_downloaded"
+	SupernodeUnknown             EventType = "supernode:unknown"
 )
 
 // EventData is a map of event data attributes using standardized keys

sdk/event/types.go

@@ -15,6 +15,9 @@ type SupernodeClient interface {
 	// HealthCheck performs a health check on the supernode
 	HealthCheck(ctx context.Context) (*grpc_health_v1.HealthCheckResponse, error)
 
+	// Download downloads the cascade action file
+	Download(ctx context.Context, in *supernodeservice.CascadeSupernodeDownloadRequest, opts ...grpc.CallOption) (*supernodeservice.CascadeSupernodeDownloadResponse, error)
+
 	// Close releases resources used by the client
 	Close(ctx context.Context) error
 }