go/worker/executor: Submit failure indicating commitment in case of s…

…torage failure

go/worker/compute/executor/committee/node.go

@@ -1094,7 +1094,7 @@ func (n *Node) proposeBatchLocked(processedBatch *processedBatch) {
 
 	// Commit I/O and state write logs to storage.
 	start := time.Now()
-	err := func() error {
+	storageErr := func() error {
 		span, ctx := tracing.StartSpanWithContext(n.ctx, "Apply(io, state)",
 			opentracing.ChildOf(state.batch.spanCtx),
 		)
@@ -1167,9 +1167,11 @@ func (n *Node) proposeBatchLocked(processedBatch *processedBatch) {
 	}()
 	storageCommitLatency.With(n.getMetricLabels()).Observe(time.Since(start).Seconds())
 
-	if err != nil {
-		n.abortBatchLocked(err)
-		return
+	if storageErr != nil {
+		n.logger.Error("storage failure, sibmitting failure indicating commitment",
+			"err", storageErr,
+		)
+		proposedResults.Failure = commitment.FailureStorageUnavailable
 	}
 
 	// Sign the commitment and submit.
@@ -1197,11 +1199,17 @@ func (n *Node) proposeBatchLocked(processedBatch *processedBatch) {
 	}()
 
 	// TODO: Add crash point.
-	n.transitionLocked(StateWaitingForFinalize{
-		batchStartTime: state.batchStartTime,
-		raw:            processedBatch.raw,
-		proposedIORoot: proposedResults.Header.IORoot,
-	})
+
+	switch storageErr {
+	case nil:
+		n.transitionLocked(StateWaitingForFinalize{
+			batchStartTime: state.batchStartTime,
+			raw:            processedBatch.raw,
+			proposedIORoot: proposedResults.Header.IORoot,
+		})
+	default:
+		n.abortBatchLocked(err)
+	}
 
 	crash.Here(crashPointBatchProposeAfter)
 }