Page MenuHomePhabricator
Differential D3808

[services] Tunnelbroker - Message expiration fulfill moved to database manager's insert
ClosedPublic
Actions

Authored by • max on Apr 21 2022, 8:29 AM.
Tags
None
Referenced Files
Unknown Object (File)
Sun, Mar 16, 9:43 AM
Unknown Object (File)
Sat, Mar 15, 11:59 PM
Unknown Object (File)
Thu, Mar 13, 4:24 AM
Unknown Object (File)
Thu, Mar 13, 4:24 AM
Unknown Object (File)
Thu, Mar 13, 4:24 AM
Unknown Object (File)
Thu, Mar 13, 4:24 AM
Unknown Object (File)
Thu, Mar 13, 4:24 AM
Unknown Object (File)
Thu, Mar 13, 4:24 AM
View All Files
Subscribers
• abosh
• adrian
ashoat
atul
• benschac
• karol
tomek

Details

Reviewers
• karol
ashoat
Commits
rCOMM41dc63dfca2d: [services] Tunnelbroker - Message expiration fulfill moved to database…
Summary

Removing the expire from the constructor because we will fulfill it at the DB insertion phase at DatabaseManager's insert.

Constant with the max message TTL in the database was added.

Test Plan

Run yarn test-tunnelbroker-service all tests are run and succeed.

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Event Timeline

• max created this revision.Apr 21 2022, 8:29 AM
Herald added subscribers: • abosh, • benschac, • karol and 4 others. · View Herald TranscriptApr 21 2022, 8:30 AM
• max edited the test plan for this revision. (Show Details)Apr 21 2022, 8:31 AM
Harbormaster completed remote builds in B8405: Diff 11729.Apr 21 2022, 8:35 AM
• max requested review of this revision.Apr 21 2022, 8:35 AM
Herald added a reviewer: • karol. · View Herald TranscriptApr 21 2022, 8:35 AM
• max added a reviewer: tomek.Apr 21 2022, 8:41 AM
• max added a child revision: D3833: [services] Tunnelbroker - Put and get messages from database in grpc handler.Apr 25 2022, 7:20 AM
• karol accepted this revision.Apr 25 2022, 7:28 AM
• karol added inline comments.
services/tunnelbroker/src/Constants.h
51 ↗(On Diff #11729)

This is a bit unclear to me

tomek requested changes to this revision.Apr 26 2022, 4:29 AM
tomek added inline comments.
services/tunnelbroker/src/Constants.h
51 ↗(On Diff #11729)

Agree. What is the meaning of 10 at the end?

It also looks like we were using milliseconds for other TTLs. So shouldn't this be 300 * 24 * 60 * 60 * 1000? (we also need to verify if it works correctly, as the value can't fit in 32 bits).

This revision now requires changes to proceed.Apr 26 2022, 4:29 AM
• max planned changes to this revision.May 2 2022, 6:45 AM
• max updated this revision to Diff 12204.May 4 2022, 5:19 AM

Rebase on master. Multiplication of the MESSAGE_RECORD_TTL was changed due to the comment.

• max planned changes to this revision.May 4 2022, 5:20 AM
Harbormaster completed remote builds in B8758: Diff 12204.May 4 2022, 6:38 AM
• max updated this revision to Diff 12219.May 4 2022, 6:38 AM

The database test was updated according to the changes.

• max planned changes to this revision.May 4 2022, 6:39 AM
• max marked 2 inline comments as done.May 4 2022, 6:46 AM
• max added inline comments.
services/tunnelbroker/src/Constants.h
51 ↗(On Diff #11729)

This is a bit unclear to me

That change makes sense to me. I've updated with these changes.

51 ↗(On Diff #11729)

Agree. What is the meaning of 10 at the end?

It also looks like we were using milliseconds for other TTLs.

This TTL is used in the DynamoDB TTL index to clear the old messages by the database itself.
We must use seconds (not milliseconds) from the Unix epoch for this purpose according to the DynamoDB documentation.

• max requested review of this revision.May 4 2022, 6:47 AM
• max marked an inline comment as done.
• max added inline comments.May 4 2022, 6:52 AM
services/tunnelbroker/src/Constants.h
30 ↗(On Diff #12219)

This description comment was updated to be more clear, because we added one more TTL and we need to distinguish them between each other.

Harbormaster completed remote builds in B8772: Diff 12219.May 4 2022, 7:39 AM
• max added a child revision: D3914: [services] Tunnelbroker - Wrap gRPC Get message response into a function in Tunnelbroker.May 4 2022, 8:11 AM
ashoat accepted this revision.May 4 2022, 9:07 AM
ashoat removed a reviewer: tomek.
This revision is now accepted and ready to land.May 4 2022, 9:07 AM
• max updated this revision to Diff 12237.May 4 2022, 11:46 AM

Rebase on master.

Harbormaster completed remote builds in B8789: Diff 12237.May 4 2022, 11:50 AM
Closed by commit rCOMM41dc63dfca2d: [services] Tunnelbroker - Message expiration fulfill moved to database… (authored by • max). · Explain WhyMay 4 2022, 11:54 AM
This revision was automatically updated to reflect the committed changes.
• max added a commit: rCOMM41dc63dfca2d: [services] Tunnelbroker - Message expiration fulfill moved to database….
• max removed a child revision: D3833: [services] Tunnelbroker - Put and get messages from database in grpc handler.May 23 2022, 6:20 AM

Revision Contents
Changeset List

PathSize
services/
tunnelbroker/
src/
4 lines
Database/
3 lines
3 lines
7 lines
test/
25 lines

Diff 12240

View Options

services/tunnelbroker/src/Constants.h

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/tunnelbroker/src/Database/DatabaseManager.cpp

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/tunnelbroker/src/Database/MessageItem.h

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/tunnelbroker/src/Database/MessageItem.cpp

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/tunnelbroker/test/DatabaseManagerTest.cpp

Loading...