Do not fail workflow task if an error occurs when marking it as completed

lib/temporal/workflow/task_processor.rb

@@ -139,6 +139,19 @@ def complete_task(commands, query_results)
           binary_checksum: binary_checksum,
           query_results: query_results
         )
+      rescue StandardError => error
+        # We rescue the error here to avoid failing the task in the process
+        # function above. One common cause of errors here is if the current
+        # workflow task is invalidated by a concurrent signal arriving while it
+        # tries to complete the workflow. In this case we do not need to and
+        # should not fail the workflow task.
+        #
+        # Not failing the workflow task will still result it being retried after
+        # a delay which is the behavior we'd want in cases like the above but
+        # also for ephemeral issues like network outages.
+        Temporal.logger.error("Unable to complete the workflow task", metadata.to_h.merge(error: error.inspect))
+
+        Temporal::ErrorHandler.handle(error, config, metadata: metadata)
       end
 
       def complete_query(result)

spec/unit/lib/temporal/workflow/task_processor_spec.rb

@@ -205,6 +205,33 @@
         end
       end
 
+      context 'when recording the workflow task complete fails' do
+        let(:exception) { GRPC::InvalidArgument.new('workflow task could not be completed') }
+
+        before { allow(connection).to receive(:respond_workflow_task_completed).and_raise(exception) }
+
+        it 'does not try to fail the workflow task' do
+          subject.process
+
+          expect(connection).to_not have_received(:respond_workflow_task_failed)
+        end
+
+        it 'calls error_handlers' do
+          reported_error = nil
+          reported_metadata = nil
+
+          config.on_error do |error, metadata: nil|
+            reported_error = error
+            reported_metadata = metadata
+          end
+
+          subject.process
+
+          expect(reported_error).to be_an_instance_of(GRPC::InvalidArgument)
+          expect(reported_metadata).to be_an_instance_of(Temporal::Metadata::WorkflowTask)
+        end
+      end
+
       context 'when workflow task raises an exception' do
         let(:exception) { StandardError.new('workflow task failed') }