end spans on cancel

src/main/kotlin/io/emeraldpay/dshackle/reader/SpannedReader.kt

@@ -33,16 +33,16 @@ class SpannedReader<K, D>(
         return reader.read(key)
             .contextWrite { ReactorSleuth.putSpanInScope(tracer, it, newSpan) }
             .doOnError {
-                newSpan.apply {
+                newSpan.tag(SPAN_ERROR, "true")
-                    tag(SPAN_ERROR, "true")
+                    .tag(SPAN_STATUS_MESSAGE, it.message)
-                    tag(SPAN_STATUS_MESSAGE, it.message)
+                    .end()
-                    end()
-                }
             }
             .doOnNext { newSpan.end() }
+            .doOnCancel {
+                newSpan.tag(SPAN_STATUS_MESSAGE, "cancelled").end()
+            }
             .switchIfEmpty {
-                newSpan.tag(SPAN_READER_RESULT, "empty result")
+                newSpan.tag(SPAN_READER_RESULT, "empty result").end()
-                newSpan.end()
                 Mono.empty()
             }
     }

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -124,6 +124,9 @@ open class NativeCall(
                         )
                     }
                     .doOnNext { callRes -> completeSpan(callRes, requestCount) }
+                    .doOnCancel {
+                        tracer.currentSpan()?.tag(SPAN_STATUS_MESSAGE, "cancel")?.end()
+                    }
                     .contextWrite { ctx -> createTracingReactorContext(ctx, requestCount, requestId, requestSpan) }
             }
     }