ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-11-26 22:15:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reprocess accepted ledgers once. Track metadata, transaction data, and affected accounts.

Browse Source 
Process to SQL database and publish from this structure.
Include number of transactions in ledger publish info.
Publish transactions in applied order.
This commit is contained in:
JoelKatz
2013-03-04 13:59:53 -08:00
parent dd398b7eb9
commit 51db2d2cd7
14 changed files with 152 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
src/cpp/ripple/AcceptedLedger.cpp
View File

@@ -1,13 +1,59 @@
#include "AcceptedLedger.h"
#include <boost/foreach.hpp>
TaggedCache<uint256, AcceptedLedger> AcceptedLedger::ALCache("AcceptedLedger", 4, 60);
ALTransaction::ALTransaction(uint32 seq, SerializerIterator& sit)
{
Serializer txnSer(sit.getVL());
SerializerIterator txnIt(txnSer);
mTxn = boost::make_shared<SerializedTransaction>(boost::ref(txnIt));
mMeta = boost::make_shared<TransactionMetaSet>(mTxn->getTransactionID(), seq, sit.getVL());
mRawMeta= sit.getVL();
mMeta = boost::make_shared<TransactionMetaSet>(mTxn->getTransactionID(), seq, mRawMeta);
mAffected = mMeta->getAffectedAccounts();
mResult = mMeta->getResultTER();
}
ALTransaction::ALTransaction(SerializedTransaction::ref txn, TransactionMetaSet::ref met) :
mTxn(txn), mMeta(met), mAffected(met->getAffectedAccounts())
{
mResult = mMeta->getResultTER();
}
ALTransaction::ALTransaction(SerializedTransaction::ref txn, TER result) :
mTxn(txn), mResult(result), mAffected(txn->getMentionedAccounts())
{ ; }
std::string ALTransaction::getEscMeta() const
{
assert(!mRawMeta.empty());
return sqlEscape(mRawMeta);
}
Json::Value ALTransaction::getJson(int j) const
{
Json::Value ret(Json::objectValue);
ret["transaction"] = mTxn->getJson(j);
if (mMeta)
{
ret["meta"] = mMeta->getJson(j);
ret["raw_meta"] = strHex(mRawMeta);
}
ret["result"] = transHuman(mResult);
if (!mAffected.empty())
{
Json::Value affected(Json::arrayValue);
BOOST_FOREACH(const RippleAddress& ra, mAffected)
{
affected.append(ra.humanAccountID());
}
ret["affected"] = affected;
}
return ret;
}
AcceptedLedger::AcceptedLedger(Ledger::ref ledger) : mLedger(ledger)
@@ -20,6 +66,16 @@ AcceptedLedger::AcceptedLedger(Ledger::ref ledger) : mLedger(ledger)
}
}
AcceptedLedger::pointer AcceptedLedger::makeAcceptedLedger(Ledger::ref ledger)
{
AcceptedLedger::pointer ret = ALCache.fetch(ledger->getHash());
if (ret)
return ret;
ret = AcceptedLedger::pointer(new AcceptedLedger(ledger));
ALCache.canonicalize(ledger->getHash(), ret);
return ret;
}
void AcceptedLedger::insert(const ALTransaction& at)
{
assert(mMap.find(at.getIndex()) == mMap.end());
@@ -32,4 +88,4 @@ const ALTransaction* AcceptedLedger::getTxn(int i) const
if (it == mMap.end())
return NULL;
return &it->second;
}
}

30
src/cpp/ripple/AcceptedLedger.h
View File

@@ -11,25 +11,38 @@ class ALTransaction
protected:
SerializedTransaction::pointer mTxn;
TransactionMetaSet::pointer mMeta;
TER mResult;
std::vector<RippleAddress> mAffected;
std::vector<unsigned char> mRawMeta;
public:
ALTransaction(uint32 ledgerSeq, SerializerIterator& sit);
ALTransaction(SerializedTransaction::ref, TransactionMetaSet::ref);
ALTransaction(SerializedTransaction::ref, TER result);
SerializedTransaction::ref getTxn() const { return mTxn; }
TransactionMetaSet::ref getMeta() const { return mMeta; }
const std::vector<RippleAddress>& getAffected() const { return mAffected; }
int getIndex() const { return mMeta->getIndex(); }
TER getResult() const { return mMeta->getResultTER(); }
uint256 getTransactionID() const { return mTxn->getTransactionID(); }
TransactionType getTxnType() const { return mTxn->getTxnType(); }
TER getResult() const { return mResult; }
bool isApplied() const { return !!mMeta; }
int getIndex() const { return mMeta ? mMeta->getIndex() : 0; }
std::string getEscMeta() const;
Json::Value getJson(int) const;
};
class AcceptedLedger
{
public:
typedef std::map<int, ALTransaction> map_t;
typedef map_t::value_type value_type;
typedef map_t::const_iterator const_iterator;
typedef boost::shared_ptr<AcceptedLedger> pointer;
typedef const pointer& ret;
typedef std::map<int, ALTransaction> map_t;
typedef map_t::value_type value_type;
typedef map_t::const_iterator const_iterator;
protected:
Ledger::pointer mLedger;
@@ -37,9 +50,14 @@ protected:
void insert(const ALTransaction&);
public:
static TaggedCache<uint256, AcceptedLedger> ALCache;
AcceptedLedger(Ledger::ref ledger);
public:
static pointer makeAcceptedLedger(Ledger::ref ledger);
static void sweep() { ALCache.sweep(); }
Ledger::ref getLedger() const { return mLedger; }
const map_t& getMap() const { return mMap; }

2
src/cpp/ripple/Application.cpp
View File

@@ -1,5 +1,6 @@
#include "Application.h"
#include "AcceptedLedger.h"
#include "Config.h"
#include "PeerDoor.h"
#include "RPCDoor.h"
@@ -303,6 +304,7 @@ void Application::sweep()
mValidations.sweep();
getMasterLedgerAcquire().sweep();
mSLECache.sweep();
AcceptedLedger::sweep();
mSweepTimer.expires_from_now(boost::posix_time::seconds(theConfig.getSize(siSweepInterval)));
mSweepTimer.async_wait(boost::bind(&Application::sweep, this));
}

38
src/cpp/ripple/Ledger.cpp
View File

@@ -428,6 +428,8 @@ void Ledger::saveAcceptedLedger(bool fromConsensus, LoadEvent::pointer event)
addRaw(s);
theApp->getHashedObjectStore().store(hotLEDGER, mLedgerSeq, s.peekData(), mHash);
AcceptedLedger::pointer aLedger = AcceptedLedger::makeAcceptedLedger(shared_from_this());
{
{
ScopedLock sl(theApp->getLedgerDB()->getDBLock());
@@ -435,33 +437,22 @@ void Ledger::saveAcceptedLedger(bool fromConsensus, LoadEvent::pointer event)
theApp->getLedgerDB()->getDB()->executeSQL(boost::str(deleteLedger % mLedgerSeq));
}
SHAMap& txSet = *peekTransactionMap();
Database *db = theApp->getTxnDB()->getDB();
{
ScopedLock dbLock(theApp->getTxnDB()->getDBLock());
db->executeSQL("BEGIN TRANSACTION;");
SHAMapTreeNode::TNType type;
for (SHAMapItem::pointer item = txSet.peekFirstItem(type); !!item;
item = txSet.peekNextItem(item->getTag(), type))
BOOST_FOREACH(const AcceptedLedger::value_type& vt, aLedger->getMap())
{
assert(type == SHAMapTreeNode::tnTRANSACTION_MD);
SerializerIterator sit(item->peekSerializer());
Serializer rawTxn(sit.getVL());
Serializer rawMeta(sit.getVL());
std::string escMeta(sqlEscape(rawMeta.peekData()));
SerializerIterator txnIt(rawTxn);
SerializedTransaction txn(txnIt);
assert(txn.getTransactionID() == item->getTag());
TransactionMetaSet meta(item->getTag(), mLedgerSeq, rawMeta.peekData());
theApp->getMasterTransaction().inLedger(item->getTag(), mLedgerSeq);
cLog(lsTRACE) << "Saving: " << vt.second.getJson(0);
uint256 txID = vt.second.getTransactionID();
theApp->getMasterTransaction().inLedger(txID, mLedgerSeq);
// Make sure transaction is in AccountTransactions.
if (!SQL_EXISTS(db, boost::str(AcctTransExists % item->getTag().GetHex())))
if (!SQL_EXISTS(db, boost::str(AcctTransExists % txID.GetHex())))
{
// Transaction not in AccountTransactions
const std::vector<RippleAddress> accts = meta.getAffectedAccounts();
const std::vector<RippleAddress>& accts = vt.second.getAffected();
if (!accts.empty())
{
@@ -476,7 +467,7 @@ void Ledger::saveAcceptedLedger(bool fromConsensus, LoadEvent::pointer event)
sql += "('";
first = false;
}
sql += txn.getTransactionID().GetHex();
sql += txID.GetHex();
sql += "','";
sql += it->humanAccountID();
sql += "',";
@@ -491,20 +482,21 @@ void Ledger::saveAcceptedLedger(bool fromConsensus, LoadEvent::pointer event)
cLog(lsWARNING) << "Transaction in ledger " << mLedgerSeq << " affects no accounts";
}
if (SQL_EXISTS(db, boost::str(transExists % txn.getTransactionID().GetHex())))
if (SQL_EXISTS(db, boost::str(transExists % txID.GetHex())))
{
// In Transactions, update LedgerSeq, metadata and Status.
db->executeSQL(boost::str(updateTx
% getLedgerSeq()
% TXN_SQL_VALIDATED
% escMeta
% txn.getTransactionID().GetHex()));
% vt.second.getEscMeta()
% txID.GetHex()));
}
else
{
// Not in Transactions, insert the whole thing..
db->executeSQL(
txn.getMetaSQLInsertHeader() + txn.getMetaSQL(getLedgerSeq(), escMeta) + ";");
SerializedTransaction::getMetaSQLInsertHeader() +
vt.second.getTxn()->getMetaSQL(getLedgerSeq(), vt.second.getEscMeta()) + ";");
}
}
db->executeSQL("COMMIT TRANSACTION;");

8
src/cpp/ripple/LedgerMaster.cpp
View File

@@ -119,11 +119,11 @@ Ledger::pointer LedgerMaster::closeLedger(bool recover)
return closingLedger;
}
TER LedgerMaster::doTransaction(const SerializedTransaction& txn, TransactionEngineParams params, bool& didApply)
TER LedgerMaster::doTransaction(SerializedTransaction::ref txn, TransactionEngineParams params, bool& didApply)
{
TER result = mEngine.applyTransaction(txn, params, didApply);
// CHECKME: Should we call this even on gross failures?
theApp->getOPs().pubProposedTransaction(mEngine.getLedger(), txn, result);
TER result = mEngine.applyTransaction(*txn, params, didApply);
// if (didApply)
theApp->getOPs().pubProposedTransaction(mEngine.getLedger(), txn, result);
return result;
}

2
src/cpp/ripple/LedgerMaster.h
View File

@@ -75,7 +75,7 @@ public:
// The published ledger is the last fully validated ledger
Ledger::ref getValidatedLedger() { return mPubLedger; }
TER doTransaction(const SerializedTransaction& txn, TransactionEngineParams params, bool& didApply);
TER doTransaction(SerializedTransaction::ref txn, TransactionEngineParams params, bool& didApply);
void pushLedger(Ledger::ref newLedger);
void pushLedger(Ledger::ref newLCL, Ledger::ref newOL, bool fromConsensus);

60
src/cpp/ripple/NetworkOPs.cpp
View File

@@ -267,7 +267,7 @@ void NetworkOPs::runTransactionQueue()
assert(dbtx);
bool didApply;
TER r = mLedgerMaster->doTransaction(*dbtx->getSTransaction(),
TER r = mLedgerMaster->doTransaction(dbtx->getSTransaction(),
tapOPEN_LEDGER | tapNO_CHECK_SIGN, didApply);
dbtx->setResult(r);
@@ -352,7 +352,7 @@ Transaction::pointer NetworkOPs::processTransaction(Transaction::pointer trans,
boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock());
Transaction::pointer dbtx = theApp->getMasterTransaction().fetch(trans->getID(), true);
bool didApply;
TER r = mLedgerMaster->doTransaction(*trans->getSTransaction(), tapOPEN_LEDGER | tapNO_CHECK_SIGN, didApply);
TER r = mLedgerMaster->doTransaction(trans->getSTransaction(), tapOPEN_LEDGER | tapNO_CHECK_SIGN, didApply);
trans->setResult(r);
if (isTemMalformed(r)) // malformed, cache bad
@@ -1289,9 +1289,9 @@ Json::Value NetworkOPs::pubBootstrapAccountInfo(Ledger::ref lpAccepted, const Ri
return jvObj;
}
void NetworkOPs::pubProposedTransaction(Ledger::ref lpCurrent, const SerializedTransaction& stTxn, TER terResult)
void NetworkOPs::pubProposedTransaction(Ledger::ref lpCurrent, SerializedTransaction::ref stTxn, TER terResult)
{
Json::Value jvObj = transJson(stTxn, terResult, false, lpCurrent, "transaction");
Json::Value jvObj = transJson(*stTxn, terResult, false, lpCurrent, "transaction");
{
boost::recursive_mutex::scoped_lock sl(mMonitorLock);
@@ -1308,15 +1308,19 @@ void NetworkOPs::pubProposedTransaction(Ledger::ref lpCurrent, const SerializedT
it = mSubRTTransactions.erase(it);
}
}
TransactionMetaSet::pointer ret;
pubAccountTransaction(lpCurrent,stTxn,terResult,false,ret);
ALTransaction alt(stTxn, terResult);
cLog(lsTRACE) << "pubProposed: " << alt.getJson(0);
pubAccountTransaction(lpCurrent, ALTransaction(stTxn, terResult));
}
void NetworkOPs::pubLedger(Ledger::ref lpAccepted)
void NetworkOPs::pubLedger(Ledger::ref accepted)
{
// Ledgers are published only when they acquire sufficient validations
// Holes are filled across connection loss or other catastrophe
AcceptedLedger::pointer alpAccepted = AcceptedLedger::makeAcceptedLedger(accepted);
Ledger::ref lpAccepted = alpAccepted->getLedger();
{
boost::recursive_mutex::scoped_lock sl(mMonitorLock);
@@ -1334,6 +1338,8 @@ void NetworkOPs::pubLedger(Ledger::ref lpAccepted)
jvObj["reserve_base"] = Json::UInt(lpAccepted->getReserve(0));
jvObj["reserve_inc"] = Json::UInt(lpAccepted->getReserveInc());
jvObj["txn_count"] = Json::UInt(alpAccepted->getTxnCount());
NetworkOPs::subMapType::const_iterator it = mSubLedger.begin();
while (it != mSubLedger.end())
{
@@ -1352,21 +1358,10 @@ void NetworkOPs::pubLedger(Ledger::ref lpAccepted)
// Don't lock since pubAcceptedTransaction is locking.
if (!mSubTransactions.empty() || !mSubRTTransactions.empty() || !mSubAccount.empty() || !mSubRTAccount.empty())
{
SHAMap& txSet = *lpAccepted->peekTransactionMap();
for (SHAMapItem::pointer item = txSet.peekFirstItem(); !!item; item = txSet.peekNextItem(item->getTag()))
BOOST_FOREACH(const AcceptedLedger::value_type& vt, alpAccepted->getMap())
{
SerializerIterator it(item->peekSerializer());
// OPTIMIZEME: Could get transaction from txn master, but still must call getVL
Serializer txnSer(it.getVL());
SerializerIterator txnIt(txnSer);
SerializedTransaction stTxn(txnIt);
TransactionMetaSet::pointer meta = boost::make_shared<TransactionMetaSet>(
stTxn.getTransactionID(), lpAccepted->getLedgerSeq(), it.getVL());
pubAcceptedTransaction(lpAccepted, stTxn, meta->getResultTER(), meta);
cLog(lsTRACE) << "pubAccepted: " << vt.second.getJson(0);
pubAcceptedTransaction(lpAccepted, vt.second);
}
}
}
@@ -1407,11 +1402,10 @@ Json::Value NetworkOPs::transJson(const SerializedTransaction& stTxn, TER terRes
return jvObj;
}
void NetworkOPs::pubAcceptedTransaction(Ledger::ref lpCurrent, const SerializedTransaction& stTxn, TER terResult,TransactionMetaSet::pointer& meta)
void NetworkOPs::pubAcceptedTransaction(Ledger::ref alAccepted, const ALTransaction& alTx)
{
Json::Value jvObj = transJson(stTxn, terResult, true, lpCurrent, "transaction");
if (meta) jvObj["meta"] = meta->getJson(0);
Json::Value jvObj = transJson(*alTx.getTxn(), alTx.getResult(), true, alAccepted, "transaction");
jvObj["meta"] = alTx.getMeta()->getJson(0);
{
boost::recursive_mutex::scoped_lock sl(mMonitorLock);
@@ -1442,14 +1436,14 @@ void NetworkOPs::pubAcceptedTransaction(Ledger::ref lpCurrent, const SerializedT
it = mSubRTTransactions.erase(it);
}
}
theApp->getOrderBookDB().processTxn(stTxn, terResult, meta, jvObj);
pubAccountTransaction(lpCurrent, stTxn, terResult, true, meta);
theApp->getOrderBookDB().processTxn(alAccepted, alTx, jvObj);
pubAccountTransaction(alAccepted, alTx);
}
void NetworkOPs::pubAccountTransaction(Ledger::ref lpCurrent, const SerializedTransaction& stTxn, TER terResult, bool bAccepted, TransactionMetaSet::pointer& meta)
void NetworkOPs::pubAccountTransaction(Ledger::ref lpCurrent, const ALTransaction& alTx)
{
boost::unordered_set<InfoSub::pointer> notify;
bool bAccepted = alTx.isApplied();
int iProposed = 0;
int iAccepted = 0;
@@ -1460,8 +1454,7 @@ void NetworkOPs::pubAccountTransaction(Ledger::ref lpCurrent, const SerializedTr
if (!mSubAccount.empty() || (!mSubRTAccount.empty()) )
{
std::vector<RippleAddress> accounts = meta ? meta->getAffectedAccounts() : stTxn.getMentionedAccounts();
BOOST_FOREACH(const RippleAddress& affectedAccount, accounts)
BOOST_FOREACH(const RippleAddress& affectedAccount, alTx.getAffected())
{
subInfoMapIterator simiIt = mSubRTAccount.find(affectedAccount.getAccountID());
@@ -1510,9 +1503,10 @@ void NetworkOPs::pubAccountTransaction(Ledger::ref lpCurrent, const SerializedTr
if (!notify.empty())
{
Json::Value jvObj = transJson(stTxn, terResult, bAccepted, lpCurrent, "account");
Json::Value jvObj = transJson(*alTx.getTxn(), alTx.getResult(), bAccepted, lpCurrent, "account");
if (meta) jvObj["meta"] = meta->getJson(0);
if (alTx.isApplied())
jvObj["meta"] = alTx.getMeta()->getJson(0);
BOOST_FOREACH(InfoSub::ref isrListener, notify)
{

7
src/cpp/ripple/NetworkOPs.h
View File

@@ -14,6 +14,7 @@
#include "LedgerAcquire.h"
#include "LedgerProposal.h"
#include "JobQueue.h"
#include "AcceptedLedger.h"
// Operations that clients may wish to perform against the network
// Master operational handler, server sequencer, network tracker
@@ -140,8 +141,8 @@ protected:
Json::Value pubBootstrapAccountInfo(Ledger::ref lpAccepted, const RippleAddress& naAccountID);
void pubAcceptedTransaction(Ledger::ref lpCurrent, const SerializedTransaction& stTxn, TER terResult,TransactionMetaSet::pointer& meta);
void pubAccountTransaction(Ledger::ref lpCurrent, const SerializedTransaction& stTxn, TER terResult,bool accepted,TransactionMetaSet::pointer& meta);
void pubAcceptedTransaction(Ledger::ref alAccepted, const ALTransaction& alTransaction);
void pubAccountTransaction(Ledger::ref lpCurrent, const ALTransaction& alTransaction);
void pubServer();
@@ -306,7 +307,7 @@ public:
// Monitoring: publisher side
//
void pubLedger(Ledger::ref lpAccepted);
void pubProposedTransaction(Ledger::ref lpCurrent, const SerializedTransaction& stTxn, TER terResult);
void pubProposedTransaction(Ledger::ref lpCurrent, SerializedTransaction::ref stTxn, TER terResult);
//

7
src/cpp/ripple/OrderBookDB.cpp
View File

@@ -187,15 +187,14 @@ BookListeners::pointer OrderBookDB::getBookListeners(const uint160& currencyIn,
*/
// Based on the meta, send the meta to the streams that are listening
// We need to determine which streams a given meta effects
void OrderBookDB::processTxn(const SerializedTransaction& stTxn, TER terResult,
TransactionMetaSet::pointer& meta, Json::Value& jvObj)
void OrderBookDB::processTxn(Ledger::ref ledger, const ALTransaction& alTx, Json::Value& jvObj)
{
boost::recursive_mutex::scoped_lock sl(mLock);
if (terResult == tesSUCCESS)
if (alTx.getResult() == tesSUCCESS)
{
// check if this is an offer or an offer cancel or a payment that consumes an offer
//check to see what the meta looks like
BOOST_FOREACH(STObject& node, meta->getNodes())
BOOST_FOREACH(STObject& node, alTx.getMeta()->getNodes())
{
try
{

3
src/cpp/ripple/OrderBookDB.h
View File

@@ -6,6 +6,7 @@
#include <boost/unordered_map.hpp>
#include "Ledger.h"
#include "AcceptedLedger.h"
#include "OrderBook.h"
@@ -66,7 +67,7 @@ public:
const uint160& issuerIn, const uint160& issuerOut);
// see if this txn effects any orderbook
void processTxn(const SerializedTransaction& stTxn, TER terResult,TransactionMetaSet::pointer& meta,Json::Value& jvObj);
void processTxn(Ledger::ref ledger, const ALTransaction& alTx, Json::Value& jvObj);
};