guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore/s3: resolve completion status via the transfer status callback

Browse Source 
Completion status can be updated in two ways:

- progress callback
- transfer status update callback (esp. important in multipart
transfers)

It's guaranteed that one of the two will provide progress for our
promise.

This fixes #945.

Change-Id: Iac9e92df34f82fbc1facae7a411ccabf1688da52
Signed-off-by: Raito Bezarius <raito@lix.systems>
This commit is contained in:
Raito Bezarius
2025-11-01 21:17:26 +01:00
parent 91867941fa
commit ccf196d7f4
1 changed files with 35 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/s3-binary-cache-store.cc
+35 -17
View File
@@ -312,6 +312,28 @@ struct S3BinaryCacheStoreConfig final : BinaryCacheStoreConfig
}
};
namespace {
struct TransferContext : Aws::Client::AsyncCallerContext
{
FulfillerWrapper<void> signal;
explicit TransferContext(kj::Own<kj::CrossThreadPromiseFulfiller<void>> signal)
: signal(std::move(signal))
{
}
};
}
static void notifyIfTransferFinished(const std::shared_ptr<const TransferHandle> & transferHandle)
{
if (transferHandle->GetStatus() != TransferStatus::NOT_STARTED
&& transferHandle->GetStatus() != TransferStatus::IN_PROGRESS)
{
auto context =
std::static_pointer_cast<const TransferContext>(transferHandle->GetContext());
context->signal.fulfill();
}
}
struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
{
S3BinaryCacheStoreConfig config_;
@@ -432,15 +454,6 @@ struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
static std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor>
executor = std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(maxThreads);
struct TransferContext : Aws::Client::AsyncCallerContext
{
FulfillerWrapper<void> signal;
explicit TransferContext(kj::Own<kj::CrossThreadPromiseFulfiller<void>> signal)
: signal(std::move(signal))
{
}
};
std::call_once(transferManagerCreated, [&]()
{
if (config().multipartUpload) {
@@ -458,14 +471,19 @@ struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
transferHandle->GetBytesTransferred(),
transferHandle->GetBytesTotalSize());
if (transferHandle->GetStatus() != TransferStatus::NOT_STARTED
&& transferHandle->GetStatus() != TransferStatus::IN_PROGRESS)
{
auto context = std::static_pointer_cast<const TransferContext>(
transferHandle->GetContext()
);
context->signal.fulfill();
}
notifyIfTransferFinished(transferHandle);
};
transferConfig.transferStatusUpdatedCallback =
[](const TransferManager * transferManager,
const std::shared_ptr<const TransferHandle> & transferHandle) {
debug(
"transfer ('%s') status updated to '%s'",
transferHandle->GetKey(),
transferHandle->GetStatus()
);
notifyIfTransferFinished(transferHandle);
};
transferManager = TransferManager::Create(transferConfig);