- Change error handling so that both synchonous and asynchronous errors can be reported back from bucket executor.

- Treat remapping as an error.
  - For lidspace compaction job iterator is reset and will be recreated on next invocation.
  - For bucketmove th ebucket is rechecked and either discarded or restarted.

4 files changed, 23 insertions, 16 deletions

diff --git a/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.cpp b/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.cpp
index b76787736ac..d3e765d3d56 100644
--- a/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.cpp
+++ b/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.cpp
@@ -22,9 +22,9 @@ DummyBucketExecutor::~DummyBucketExecutor() {
     sync();
 }
 
-std::unique_ptr<BucketTask>
+void
 DummyBucketExecutor::execute(const Bucket & bucket, std::unique_ptr<BucketTask> task) {
-    _executor->execute(makeLambdaTask([this, bucket, bucketTask=std::move(task)]() {
+    auto failed = _executor->execute(makeLambdaTask([this, bucket, bucketTask=std::move(task)]() {
         {
             std::unique_lock guard(_lock);
             // Use contains when dropping support for gcc 8.
@@ -41,7 +41,9 @@ DummyBucketExecutor::execute(const Bucket & bucket, std::unique_ptr<BucketTask>
             _cond.notify_all();
         }));
     }));
-    return task;
+    if (failed) {
+        failed->run();
+    }
 }
 
 void
diff --git a/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.h b/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.h
index 2479787eb31..a3cf146a0f4 100644
--- a/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.h
+++ b/persistence/src/vespa/persistence/dummyimpl/dummy_bucket_executor.h
@@ -17,7 +17,7 @@ class DummyBucketExecutor : public BucketExecutor {
 public:
     DummyBucketExecutor(size_t numExecutors);
     ~DummyBucketExecutor() override;
-    std::unique_ptr<BucketTask> execute(const Bucket & bucket, std::unique_ptr<BucketTask> task) override;
+    void execute(const Bucket & bucket, std::unique_ptr<BucketTask> task) override;
     void sync();
 private:
     std::unique_ptr<vespalib::SyncableThreadExecutor> _executor;
diff --git a/persistence/src/vespa/persistence/spi/bucket_tasks.h b/persistence/src/vespa/persistence/spi/bucket_tasks.h
index 0b9c283817d..9468dd09dc3 100644
--- a/persistence/src/vespa/persistence/spi/bucket_tasks.h
+++ b/persistence/src/vespa/persistence/spi/bucket_tasks.h
@@ -9,11 +9,12 @@ namespace storage::spi {
 /**
  * Simple Bucket task that wraps a lambda that does the job.
  */
-template<class FunctionType>
+template<class FunctionType, class FailedFunction>
 class LambdaBucketTask : public BucketTask {
 public:
-    explicit LambdaBucketTask(FunctionType &&func)
-            : _func(std::move(func))
+    explicit LambdaBucketTask(FunctionType &&func, FailedFunction &&failed)
+        : _func(std::move(func)),
+          _failed(std::move(failed))
     {}
 
     ~LambdaBucketTask() override = default;
@@ -21,16 +22,20 @@ public:
     void run(const Bucket & bucket, std::shared_ptr<vespalib::IDestructorCallback> onComplete) override {
         _func(bucket, std::move(onComplete));
     }
+    void fail(const Bucket & bucket) override {
+        _failed(bucket);
+    }
 
 private:
-    FunctionType _func;
+    FunctionType   _func;
+    FailedFunction _failed;
 };
 
-template<class FunctionType>
+template<class FunctionType, class FailedFunction>
 std::unique_ptr<BucketTask>
-makeBucketTask(FunctionType &&function) {
-    return std::make_unique<LambdaBucketTask<std::decay_t<FunctionType>>>
-    (std::forward<FunctionType>(function));
+makeBucketTask(FunctionType &&function, FailedFunction && failed) {
+    return std::make_unique<LambdaBucketTask<std::decay_t<FunctionType>, std::decay_t<FailedFunction>>>
+    (std::forward<FunctionType>(function), std::forward<FailedFunction>(failed));
 }
     
 }
diff --git a/persistence/src/vespa/persistence/spi/bucketexecutor.h b/persistence/src/vespa/persistence/spi/bucketexecutor.h
index 26884dd04ae..350376be595 100644
--- a/persistence/src/vespa/persistence/spi/bucketexecutor.h
+++ b/persistence/src/vespa/persistence/spi/bucketexecutor.h
@@ -20,16 +20,16 @@ class BucketTask {
 public:
     virtual ~BucketTask() = default;
     virtual void run(const Bucket & bucket, std::shared_ptr<vespalib::IDestructorCallback> onComplete) = 0;
+    virtual void fail(const Bucket &) = 0;
 };
 
 /**
- * Interface for running a BucketTask. If it fails the task will be returned.
- * That would normally indicate a fatal error.
- * sync() will be called during detatching to ensure the implementation is drained.
+ * Interface for running a BucketTask. If running the task fails either synchronously or asynchronously.
+ * The fail method will be invoked, either synchronously or asynchronously.
  */
 struct BucketExecutor {
     virtual ~BucketExecutor() = default;
-    virtual std::unique_ptr<BucketTask> execute(const Bucket & bucket, std::unique_ptr<BucketTask> task) = 0;
+    virtual void execute(const Bucket & bucket, std::unique_ptr<BucketTask> task) = 0;
 };
 
 }