sdk/cascade: split upload vs processing timeouts; add explicit constants; add upload/processing events; add finalize simulation failure event; detailed artefact metrics; docs

Author: mateeullahmalik

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

@@ -61,4 +61,5 @@ enum SupernodeEventType {
   ARTEFACTS_STORED = 11;
   ACTION_FINALIZED = 12;
   ARTEFACTS_DOWNLOADED = 13;
+  FINALIZE_SIMULATION_FAILED = 14;
 }

proto/supernode/action/cascade/service.proto

@@ -365,8 +365,13 @@ The SDK provides an event system to monitor task progress through event subscrip
 - `SDKTaskTxHashReceived`: Transaction hash received from supernode
 - `SDKTaskCompleted`: Task completed successfully
 - `SDKTaskFailed`: Task failed with error
- - `SDKUploadTimeout`: Upload phase exceeded time budget and was cancelled
-- `SDKProcessingTimeout`: Post-upload processing exceeded time budget and was cancelled
+ - `SDKConnectionEstablished`: Connection to supernode established
+ - `SDKUploadStarted`: Upload started (size, chunk size, est chunks)
+ - `SDKUploadCompleted`: Upload completed (size, chunks, elapsed, avg throughput)
+ - `SDKUploadFailed`: Upload failed (reason=timeout|send_error|read_error|file_open|file_stat|close_send)
+ - `SDKProcessingStarted`: Waiting for server progress/final tx hash
+ - `SDKProcessingFailed`: Processing failed (reason=stream_recv|missing_final_response)
+ - `SDKProcessingTimeout`: Processing exceeded time budget and was cancelled
 - `SDKDownloadAttempt`: Attempting to download from supernode
 - `SDKDownloadFailure`: Download attempt failed
 - `SDKOutputPathReceived`: File download path received
@@ -384,9 +389,10 @@ The SDK provides an event system to monitor task progress through event subscrip
 - `SupernodeRQIDVerified`: RaptorQ ID verified
 - `SupernodeFinalizeSimulated`: Finalize transaction simulated successfully (pre-storage)
 - `SupernodeArtefactsStored`: Artifacts stored successfully
-- `SupernodeActionFinalized`: Action processing finalized
-- `SupernodeArtefactsDownloaded`: Artifacts downloaded
-- `SupernodeUnknown`: Unknown supernode event
+ - `SupernodeActionFinalized`: Action processing finalized
+ - `SupernodeArtefactsDownloaded`: Artifacts downloaded
+ - `SupernodeFinalizeSimulationFailed`: Finalize action simulation failed
+ - `SupernodeUnknown`: Unknown supernode event
 
 Note: For backward compatibility, older supernodes may emit the finalize simulation as an `RQID_VERIFIED` event with the message `"finalize action simulation passed"`. The SDK adapter maps this to `SupernodeFinalizeSimulated` automatically.
 

sdk/README.md

@@ -90,13 +90,25 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 	if err != nil {
 		a.logger.Error(ctx, "Failed to create register stream",
 			"error", err)
+		if in.EventLogger != nil {
+			in.EventLogger(baseCtx, event.SDKUploadFailed, "upload failed | reason=stream_open", event.EventData{
+				event.KeyTaskID:   in.TaskId,
+				event.KeyActionID: in.ActionID,
+			})
+		}
 		return nil, err
 	}
 
 	// Open the file for reading
 	file, err := os.Open(in.FilePath)
 	if err != nil {
 		a.logger.Error(ctx, "Failed to open file", "filePath", in.FilePath, "error", err)
+		if in.EventLogger != nil {
+			in.EventLogger(baseCtx, event.SDKUploadFailed, "upload failed | reason=file_open", event.EventData{
+				event.KeyTaskID:   in.TaskId,
+				event.KeyActionID: in.ActionID,
+			})
+		}
 		return nil, fmt.Errorf("failed to open file: %w", err)
 	}
 	defer file.Close()
@@ -105,6 +117,12 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 	fileInfo, err := file.Stat()
 	if err != nil {
 		a.logger.Error(ctx, "Failed to get file stats", "filePath", in.FilePath, "error", err)
+		if in.EventLogger != nil {
+			in.EventLogger(baseCtx, event.SDKUploadFailed, "upload failed | reason=file_stat", event.EventData{
+				event.KeyTaskID:   in.TaskId,
+				event.KeyActionID: in.ActionID,
+			})
+		}
 		return nil, fmt.Errorf("failed to get file stats: %w", err)
 	}
 	totalBytes := fileInfo.Size()
@@ -128,14 +146,21 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 	chunkIndex := 0
 	buffer := make([]byte, chunkSize)
 
+	// Emit upload started event
+	if in.EventLogger != nil {
+		estChunks := (totalBytes + int64(chunkSize) - 1) / int64(chunkSize)
+		in.EventLogger(baseCtx, event.SDKUploadStarted,
+			fmt.Sprintf("upload started | size=%dB chunk_size=%dB est_chunks=%d", totalBytes, chunkSize, estChunks),
+			event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
+	}
+
+	uploadStart := time.Now()
+
 	// Start upload phase timer
 	uploadTimer := time.AfterFunc(cascadeUploadTimeout, func() {
 		a.logger.Error(baseCtx, "Upload phase timeout reached; cancelling stream")
 		if in.EventLogger != nil {
-			in.EventLogger(baseCtx, event.SDKUploadTimeout, "upload phase timeout", event.EventData{
-				event.KeyTaskID:   in.TaskId,
-				event.KeyActionID: in.ActionID,
-			})
+			in.EventLogger(baseCtx, event.SDKUploadFailed, "upload failed | reason=timeout", event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
 		}
 		cancel()
 	})
@@ -149,6 +174,12 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 		}
 		if err != nil {
 			a.logger.Error(ctx, "Failed to read file chunk", "chunkIndex", chunkIndex, "error", err)
+			if in.EventLogger != nil {
+				in.EventLogger(baseCtx, event.SDKUploadFailed, fmt.Sprintf("upload failed | reason=read_error chunk=%d", chunkIndex), event.EventData{
+					event.KeyTaskID:   in.TaskId,
+					event.KeyActionID: in.ActionID,
+				})
+			}
 			return nil, fmt.Errorf("failed to read file chunk: %w", err)
 		}
 
@@ -163,6 +194,12 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 
 		if err := stream.Send(chunk); err != nil {
 			a.logger.Error(ctx, "Failed to send data chunk", "chunkIndex", chunkIndex, "error", err)
+			if in.EventLogger != nil {
+				in.EventLogger(baseCtx, event.SDKUploadFailed, fmt.Sprintf("upload failed | reason=send_error chunk=%d", chunkIndex), event.EventData{
+					event.KeyTaskID:   in.TaskId,
+					event.KeyActionID: in.ActionID,
+				})
+			}
 			return nil, fmt.Errorf("failed to send chunk: %w", err)
 		}
 
@@ -186,13 +223,25 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 
 	if err := stream.Send(metadata); err != nil {
 		a.logger.Error(ctx, "Failed to send metadata", "TaskId", in.TaskId, "ActionID", in.ActionID, "error", err)
+		if in.EventLogger != nil {
+			in.EventLogger(baseCtx, event.SDKUploadFailed, "upload failed | reason=send_metadata", event.EventData{
+				event.KeyTaskID:   in.TaskId,
+				event.KeyActionID: in.ActionID,
+			})
+		}
 		return nil, fmt.Errorf("failed to send metadata: %w", err)
 	}
 
 	a.logger.Debug(ctx, "Sent metadata", "TaskId", in.TaskId, "ActionID", in.ActionID)
 
 	if err := stream.CloseSend(); err != nil {
 		a.logger.Error(ctx, "Failed to close stream and receive response", "TaskId", in.TaskId, "ActionID", in.ActionID, "error", err)
+		if in.EventLogger != nil {
+			in.EventLogger(baseCtx, event.SDKUploadFailed, "upload failed | reason=close_send", event.EventData{
+				event.KeyTaskID:   in.TaskId,
+				event.KeyActionID: in.ActionID,
+			})
+		}
 		return nil, fmt.Errorf("failed to receive response: %w", err)
 	}
 
@@ -201,14 +250,24 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 		uploadTimer.Stop()
 	}
 
+	// Emit upload completed with throughput metrics
+	if in.EventLogger != nil {
+		elapsed := time.Since(uploadStart).Seconds()
+		mb := float64(bytesRead) / (1024.0 * 1024.0)
+		avg := 0.0
+		if elapsed > 0 {
+			avg = mb / elapsed
+		}
+		in.EventLogger(baseCtx, event.SDKUploadCompleted,
+			fmt.Sprintf("upload complete | size=%dB chunks=%d elapsed=%.2fs avg=%.2fMB/s", totalBytes, chunkIndex, elapsed, avg),
+			event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
+	}
+
 	// Processing phase timer starts now (waiting for server streamed responses)
 	processingTimer := time.AfterFunc(cascadeProcessingTimeout, func() {
 		a.logger.Error(baseCtx, "Processing phase timeout reached; cancelling stream")
 		if in.EventLogger != nil {
-			in.EventLogger(baseCtx, event.SDKProcessingTimeout, "processing phase timeout", event.EventData{
-				event.KeyTaskID:   in.TaskId,
-				event.KeyActionID: in.ActionID,
-			})
+			in.EventLogger(baseCtx, event.SDKProcessingTimeout, "processing timeout", event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
 		}
 		cancel()
 	})
@@ -218,6 +277,11 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 		}
 	}()
 
+	// Emit processing started
+	if in.EventLogger != nil {
+		in.EventLogger(baseCtx, event.SDKProcessingStarted, "processing started | awaiting server progress", event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
+	}
+
 	// Handle streaming responses from supernode
 	var finalResp *cascade.RegisterResponse
 	for {
@@ -228,12 +292,16 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 		if err != nil {
 			// Distinguish timeout phase for clearer error messages
 			if phaseCtx.Err() != nil {
-				// Determine which phase we were in by whether upload finished
 				// At this point, upload is finished; classify as processing timeout/cancel
 				if phaseCtx.Err() == context.DeadlineExceeded || phaseCtx.Err() == context.Canceled {
 					return nil, fmt.Errorf("processing timed out or cancelled: %w", phaseCtx.Err())
 				}
 			}
+			if in.EventLogger != nil {
+				in.EventLogger(baseCtx, event.SDKProcessingFailed,
+					fmt.Sprintf("processing failed | reason=stream_recv error=%v", err),
+					event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
+			}
 			return nil, fmt.Errorf("failed to receive server response: %w", err)
 		}
 
@@ -268,6 +336,9 @@ func (a *cascadeAdapter) CascadeSupernodeRegister(ctx context.Context, in *Casca
 		if phaseCtx.Err() != nil {
 			return nil, fmt.Errorf("processing timed out or cancelled before final response: %w", phaseCtx.Err())
 		}
+		if in.EventLogger != nil {
+			in.EventLogger(baseCtx, event.SDKProcessingFailed, "processing failed | reason=missing_final_response", event.EventData{event.KeyTaskID: in.TaskId, event.KeyActionID: in.ActionID})
+		}
 		return nil, fmt.Errorf("no final response with tx_hash received")
 	}
 
@@ -380,7 +451,7 @@ func (a *cascadeAdapter) CascadeSupernodeDownload(
 
 // toSdkEvent converts a supernode-side enum value into an internal SDK EventType.
 func toSdkEvent(e cascade.SupernodeEventType) event.EventType {
-	switch e {
+    switch e {
 	case cascade.SupernodeEventType_ACTION_RETRIEVED:
 		return event.SupernodeActionRetrieved
 	case cascade.SupernodeEventType_ACTION_FEE_VERIFIED:
@@ -405,8 +476,10 @@ func toSdkEvent(e cascade.SupernodeEventType) event.EventType {
 		return event.SupernodeActionFinalized
 	case cascade.SupernodeEventType_ARTEFACTS_DOWNLOADED:
 		return event.SupernodeArtefactsDownloaded
-	case cascade.SupernodeEventType_FINALIZE_SIMULATED:
-		return event.SupernodeFinalizeSimulated
+    case cascade.SupernodeEventType_FINALIZE_SIMULATED:
+        return event.SupernodeFinalizeSimulated
+    case cascade.SupernodeEventType_FINALIZE_SIMULATION_FAILED:
+        return event.SupernodeFinalizeSimulationFailed
 	default:
 		return event.SupernodeUnknown
 	}

sdk/adapters/supernodeservice/adapter.go

@@ -14,18 +14,23 @@ type EventType string
 // These events are used to track the progress of tasks
 // and to notify subscribers about important changes in the system.
 const (
-	SDKTaskStarted            EventType = "sdk:started"
+    SDKTaskStarted            EventType = "sdk:started"
 	SDKSupernodesUnavailable  EventType = "sdk:supernodes_unavailable"
 	SDKSupernodesFound        EventType = "sdk:supernodes_found"
 	SDKRegistrationAttempt    EventType = "sdk:registration_attempt"
 	SDKRegistrationFailure    EventType = "sdk:registration_failure"
 	SDKRegistrationSuccessful EventType = "sdk:registration_successful"
 	SDKTaskTxHashReceived     EventType = "sdk:txhash_received"
 	SDKTaskCompleted          EventType = "sdk:completed"
-	SDKTaskFailed             EventType = "sdk:failed"
-	// Fine-grained timeouts for cascade registration
-	SDKUploadTimeout     EventType = "sdk:upload_timeout"
-	SDKProcessingTimeout EventType = "sdk:processing_timeout"
+    SDKTaskFailed             EventType = "sdk:failed"
+    SDKConnectionEstablished  EventType = "sdk:connection_established"
+    // Upload/processing phase events for cascade registration
+    SDKUploadStarted     EventType = "sdk:upload_started"
+    SDKUploadCompleted   EventType = "sdk:upload_completed"
+    SDKUploadFailed      EventType = "sdk:upload_failed"      // reason includes timeout
+    SDKProcessingStarted EventType = "sdk:processing_started"
+    SDKProcessingFailed  EventType = "sdk:processing_failed"
+    SDKProcessingTimeout EventType = "sdk:processing_timeout"
 
 	SDKDownloadAttempt    EventType = "sdk:download_attempt"
 	SDKDownloadFailure    EventType = "sdk:download_failure"
@@ -45,9 +50,10 @@ const (
 	SupernodeRQIDVerified        EventType = "supernode:rqid_verified"
 	SupernodeFinalizeSimulated   EventType = "supernode:finalize_simulated"
 	SupernodeArtefactsStored     EventType = "supernode:artefacts_stored"
-	SupernodeActionFinalized     EventType = "supernode:action_finalized"
-	SupernodeArtefactsDownloaded EventType = "supernode:artefacts_downloaded"
-	SupernodeUnknown             EventType = "supernode:unknown"
+    SupernodeActionFinalized     EventType = "supernode:action_finalized"
+    SupernodeArtefactsDownloaded EventType = "supernode:artefacts_downloaded"
+    SupernodeUnknown             EventType = "supernode:unknown"
+    SupernodeFinalizeSimulationFailed EventType = "supernode:finalize_simulation_failed"
 )
 
 // EventData is a map of event data attributes using standardized keys

sdk/event/types.go

@@ -103,15 +103,21 @@ func (t *CascadeTask) registerWithSupernodes(ctx context.Context, supernodes lum
 }
 
 func (t *CascadeTask) attemptRegistration(ctx context.Context, _ int, sn lumera.Supernode, factory *net.ClientFactory, req *supernodeservice.CascadeSupernodeRegisterRequest) error {
-	client, err := factory.CreateClient(ctx, sn)
-	if err != nil {
-		return fmt.Errorf("create client %s: %w", sn.CosmosAddress, err)
-	}
-	defer client.Close(ctx)
-
-	req.EventLogger = func(ctx context.Context, evt event.EventType, msg string, data event.EventData) {
-		t.LogEvent(ctx, evt, msg, data)
-	}
+    client, err := factory.CreateClient(ctx, sn)
+    if err != nil {
+        return fmt.Errorf("create client %s: %w", sn.CosmosAddress, err)
+    }
+    defer client.Close(ctx)
+
+    // Emit connection established event for observability
+    t.LogEvent(ctx, event.SDKConnectionEstablished, "connection established", event.EventData{
+        event.KeySupernode:        sn.GrpcEndpoint,
+        event.KeySupernodeAddress: sn.CosmosAddress,
+    })
+
+    req.EventLogger = func(ctx context.Context, evt event.EventType, msg string, data event.EventData) {
+        t.LogEvent(ctx, evt, msg, data)
+    }
 	// Use ctx directly; per-phase timers are applied inside the adapter
 	resp, err := client.RegisterCascade(ctx, req)
 	if err != nil {

sdk/task/cascade.go

@@ -3,18 +3,19 @@ package cascade
 type SupernodeEventType int
 
 const (
-	SupernodeEventTypeUNKNOWN                 SupernodeEventType = 0
-	SupernodeEventTypeActionRetrieved         SupernodeEventType = 1
-	SupernodeEventTypeActionFeeVerified       SupernodeEventType = 2
-	SupernodeEventTypeTopSupernodeCheckPassed SupernodeEventType = 3
-	SupernodeEventTypeMetadataDecoded         SupernodeEventType = 4
-	SupernodeEventTypeDataHashVerified        SupernodeEventType = 5
-	SupernodeEventTypeInputEncoded            SupernodeEventType = 6
-	SupernodeEventTypeSignatureVerified       SupernodeEventType = 7
-	SupernodeEventTypeRQIDsGenerated          SupernodeEventType = 8
-	SupernodeEventTypeRqIDsVerified           SupernodeEventType = 9
-	SupernodeEventTypeFinalizeSimulated       SupernodeEventType = 10
-	SupernodeEventTypeArtefactsStored         SupernodeEventType = 11
-	SupernodeEventTypeActionFinalized         SupernodeEventType = 12
-	SupernodeEventTypeArtefactsDownloaded     SupernodeEventType = 13
+    SupernodeEventTypeUNKNOWN                 SupernodeEventType = 0
+    SupernodeEventTypeActionRetrieved         SupernodeEventType = 1
+    SupernodeEventTypeActionFeeVerified       SupernodeEventType = 2
+    SupernodeEventTypeTopSupernodeCheckPassed SupernodeEventType = 3
+    SupernodeEventTypeMetadataDecoded         SupernodeEventType = 4
+    SupernodeEventTypeDataHashVerified        SupernodeEventType = 5
+    SupernodeEventTypeInputEncoded            SupernodeEventType = 6
+    SupernodeEventTypeSignatureVerified       SupernodeEventType = 7
+    SupernodeEventTypeRQIDsGenerated          SupernodeEventType = 8
+    SupernodeEventTypeRqIDsVerified           SupernodeEventType = 9
+    SupernodeEventTypeFinalizeSimulated       SupernodeEventType = 10
+    SupernodeEventTypeArtefactsStored         SupernodeEventType = 11
+    SupernodeEventTypeActionFinalized         SupernodeEventType = 12
+    SupernodeEventTypeArtefactsDownloaded     SupernodeEventType = 13
+    SupernodeEventTypeFinalizeSimulationFailed SupernodeEventType = 14
 )