Made iceberg order type more robust.

We were asserting that completed orders must have non zero fills, but
its easy for an algo to break that so we handle it.

src/lib/config/protos/MarketData.proto

@@ -76,6 +76,10 @@ message CMEOptions
       repeated uint32 affinity = 10;
       optional uint32 udp_socket_buffer_size = 11 [default = 33554432]; //32mb
    }
+   enum BookType {
+       BOOKTYPE_CONSOLIDATED = 0;
+       BOOKTYPE_DIRECT = 1;
+   }
    message CMEChannel {
       repeated string id = 2;
       optional string feed_engine = 3 [default = "default"];
@@ -98,6 +102,7 @@ message CMEOptions
       optional bool join_snapshot_on_gap = 32 [default = false];
 
       optional bool load_security_db = 33 [default = true];
+      optional BookType book_type = 34 [default = BOOKTYPE_CONSOLIDATED];
 
    }
    optional string onixs_license = 1 [default = "{default}"];

src/lib/md/SecurityIDRecorder.cpp

@@ -199,9 +199,18 @@ void SecurityIDRecorder::bulk_insert(const std::vector<const api::Security*>& da
    if (data.size() == 0) return;
    sqlite3_exec_nofail(_self->db.get(), "BEGIN TRANSACTION");
    std::string descriptor;
-   for (const api::Security* sec : data) {
-      descriptor = security_descriptor(*sec);
-      insert(prefix + descriptor, securitytype_of(*sec));
+   {
+      auto insert_stmt = prepare_query(_self->db, "INSERT INTO security_keys (rowid, descriptor, type) values (?, ?, ?)");
+      for (const api::Security* sec : data) {
+         descriptor = prefix + security_descriptor(*sec);
+         auto type = securitytype_of(*sec);
+
+         sqlite3_bind_int(insert_stmt.get(), 1, sec->id());
+         sqlite3_bind_text(insert_stmt.get(), 2, descriptor.c_str(), descriptor.size(), SQLITE_STATIC);
+         sqlite3_bind_int(insert_stmt.get(), 3, (int)type);
+         sqlite3_step(insert_stmt.get());
+         sqlite3_reset(insert_stmt.get());
+      }
    }
    sqlite3_exec_nofail(_self->db.get(), "COMMIT TRANSACTION");
 }

src/lib/md/relay/RelayRecorder.cpp

@@ -121,9 +121,6 @@ void record_security_keys(md::SecurityIDRecorder& idrec, const co2::MDReply& rep
          }
       }
    }
-   // Sort by id only actually works if the recording was done on a fresh machine
-   std::sort(data.begin(), data.end(), [](const Security* a, const Security* b) { return a->id() < b->id(); });
-   // We could insert garbage until data[0]->id()
    idrec.bulk_insert(data, "REPLAY: ");
 }
 #endif

src/server/sparkd/cmds/Run.cpp

@@ -8,6 +8,7 @@
 #include "lib/logging/logging.h"
 #include "lib/os/os.h"
 #include "lib/quickbus/quickbus.h"
+#include "lib/utils/Dispatcher.h"
 
 #include "lib/md/SyntheticSecurityService.h"
 #include "server/sparkd/exchange_sim/ExchangeSim.h"
@@ -53,6 +54,11 @@ int Run::execute(std::vector<std::string> const &) {
    orderman::OrderServiceServer order_service_server(sc, synthetic_service);
    support::ErrorReportServer error_report_service(sc);
    admin::AdminServiceServer admin_service(sc);
+
+   auto bg_health = std::make_shared<metrics::IntCounterMetric>("spark_background_thread");
+   utils::DispatcherTimer timer(utils::Dispatcher::background(), 5000, [=]() { bg_health->increment(); });
+   timer.start();
+   stats.publish(bg_health);
    
    std::thread([]() {
       os::set_thread_name("ExchangeSim");

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Listener.cpp

@@ -1,8 +1,8 @@
 #include "MDP3Listener.h"
 #include "lib/utils/datetime.h"
 #include <vector>
-#include "server/sparkd/tracing/bts_trace.h"
 #include "lib/utils/helper_macros.h"
+#include "lib/config/protos/MarketData.pb.h"
 
 using namespace std;
 using namespace bts::co2;
@@ -84,7 +84,8 @@ MDP3Listener::MDP3Listener(
       MDProducer& producer,
       SecurityIDRecorder& recorder,
       co2::SettlementRecorder& ,
-      std::vector<MDProducerChannel> channels) :
+      const std::vector<MDProducerChannel> channels,
+      CMEOptions::BookType bt) :
    _index(index),
    _producer(producer),
    _packet_arrived(),
@@ -94,6 +95,7 @@ MDP3Listener::MDP3Listener(
    _symbol_filter_storage(),
    _symbol_filter(),
    _channels(channels),
+   _booktype(bt),
    _concurrency_monitor(0),
    _book_updates()
 {
@@ -188,11 +190,6 @@ double double_of_price(const Price& price, int8_t display_factor) {
    return (price.mantissa() * pow10(price.exponent())) * pow10(display_factor);
 }
 
-inline PriceSize pricesize_of_price_size(const onixs::PRICE price, int8_t display_factor, Size size) {
-   PriceSize result(price.mantissa(), price.exponent() + display_factor, size);
-   return result;
-}
-
 inline datetime::LLDateTime ll_of_timestamp(const onixs::Timestamp& ts) {
    return datetime::ll_from_parts(ts.year(), (int)ts.month(), ts.day(), ts.hour(), ts.minute(), ts.second(), ts.millisecond());
 }
@@ -335,49 +332,18 @@ void MDP3Listener::onDefinition(onixs::Handler&, const onixs::Security&, const o
    _index.add_spread_legs(fsec.numeric_id, leg_exchange_ids);
 }
 
-
-//
-// Book
-void MDP3Listener::onBookUpdate(onixs::Handler&, const onixs::Security& sec, const onixs::ConsolidatedBook& book) {
-   tracepoint (bts_trace,trace_order,(char*) "onBookUpdate", 0,_latest_sequence);
-
-   auto data = _index.get_security(sec.id());
-   if (data == nullptr) return;
-   bts_assert(data->id == (uint32_t)sec.id());
-   auto display_factor = data->handle.security().display_factor;
-
-   // Put the data into the cache. Flush at the end of the packet.
-   auto key = data->handle.key();
-   auto find_iter = std::find_if(_book_updates.begin(), _book_updates.end(), [key](const BookData& entry) {
-      return key == entry.handle.key();
-   });
-   bool found = find_iter != _book_updates.end();
-   if (!found) {
-      _book_updates.emplace_back(data->handle);
-   }
-   BookData& cache = found ? *find_iter : _book_updates.back();
-
-   // Overwrite all entries every time
-   size_t bids_size = book.bids().size();
-   size_t asks_size = book.offers().size();
-   if (!data->handle.has_depth()) {
-      bids_size = min(bids_size, size_t { 1 });
-      asks_size = min(asks_size, size_t { 1 });
-   }
-   for (size_t x = 0; x < bids_size; ++x) {
-      cache.bids[x] = pricesize_of_price_size(book.bids()[x].price(), display_factor, book.bids()[x].quantity());
-   }
-   for (size_t x = 0; x < asks_size; ++x) {
-      cache.asks[x] = pricesize_of_price_size(book.offers()[x].price(), display_factor, book.offers()[x].quantity());
-   }
-   // NaN out the remaining depth
-   for (size_t x = bids_size; x < COCONUT_MAX_DEPTH; ++x) {
-      cache.bids[x] = PriceSize::NO_MARKET;
-   }
-   for (size_t x = asks_size; x < COCONUT_MAX_DEPTH; ++x) {
-      cache.asks[x] = PriceSize::NO_MARKET;
+void MDP3Listener::onBookUpdate(onixs::Handler&, const onixs::Security& sec, const onixs::DirectBook& book) {
+   if (_booktype == CMEOptions::BOOKTYPE_DIRECT) {
+      onBookUpdate(sec, book);
    }
 }
+
+void MDP3Listener::onBookUpdate(onixs::Handler&, const onixs::Security& sec, const onixs::ConsolidatedBook& book) {
+   if (_booktype == CMEOptions::BOOKTYPE_CONSOLIDATED) {
+      onBookUpdate(sec, book);
+   }
+}
+
 //
 // Trades
 

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Listener.h

@@ -3,6 +3,7 @@
 #include "lib/utils/bts_no_copy.h"
 #include "lib/md/coconut2.h"
 #include "CMESecurityIndex.h"
+#include "server/sparkd/tracing/bts_trace.h"
 
 #include "onixs.h"
 #include <tbb/concurrent_queue.h>
@@ -13,13 +14,23 @@
 namespace bts {
 namespace md {
 
+inline PriceSize pricesize_of_price_size(const onixs::PRICE price, int8_t display_factor, Size size) {
+   PriceSize result(price.mantissa(), price.exponent() + display_factor, size);
+   return result;
+}
+
 class MDP3Listener :
    public onixs::HandlerListener,
    public onixs::MarketDataListener,
    public onixs::SecurityListener
 {
 public:
-   MDP3Listener(CMESecurityIndex& index, MDProducer&, SecurityIDRecorder&, co2::SettlementRecorder&, std::vector<MDProducerChannel>);
+   MDP3Listener(CMESecurityIndex& index, 
+                MDProducer&, 
+                SecurityIDRecorder&, 
+                co2::SettlementRecorder&, 
+                const std::vector<MDProducerChannel>,
+                CMEOptions::BookType);
    virtual ~MDP3Listener();
 
    //
@@ -51,10 +62,52 @@ public:
    //
    // Book
    // void onBookUpdate(onixs::Handler&, const onixs::Security&, const onixs::MboBook&) override;
-   // void onBookUpdate(onixs::Handler&, const onixs::Security&, const onixs::DirectBook&) override;
+   void onBookUpdate(onixs::Handler&, const onixs::Security&, const onixs::DirectBook&) override;
    // void onBookUpdate(onixs::Handler&, const onixs::Security&, const onixs::ImpliedBook&) override;
    void onBookUpdate(onixs::Handler&, const onixs::Security&, const onixs::ConsolidatedBook&) override;
 
+   template <class TBook>
+   void onBookUpdate(const onixs::Security& sec, const TBook& book) {
+      tracepoint (bts_trace,trace_order,(char*) "onBookUpdate", 0,_latest_sequence);
+
+      auto data = _index.get_security(sec.id());
+      if (data == nullptr) return;
+      bts_assert(data->id == (uint32_t)sec.id());
+      auto display_factor = data->handle.security().display_factor;
+
+      // Put the data into the cache. Flush at the end of the packet.
+      auto key = data->handle.key();
+      auto find_iter = std::find_if(_book_updates.begin(), _book_updates.end(), [key](const BookData& entry) {
+         return key == entry.handle.key();
+      });
+      bool found = find_iter != _book_updates.end();
+      if (!found) {
+         _book_updates.emplace_back(data->handle);
+      }
+      BookData& cache = found ? *find_iter : _book_updates.back();
+
+      // Overwrite all entries every time
+      size_t bids_size = book.bids().size();
+      size_t asks_size = book.offers().size();
+      if (!data->handle.has_depth()) {
+         bids_size = std::min(bids_size, size_t { 1 });
+         asks_size = std::min(asks_size, size_t { 1 });
+      }
+      for (size_t x = 0; x < bids_size; ++x) {
+         cache.bids[x] = pricesize_of_price_size(book.bids()[x].price(), display_factor, book.bids()[x].quantity());
+      }
+      for (size_t x = 0; x < asks_size; ++x) {
+         cache.asks[x] = pricesize_of_price_size(book.offers()[x].price(), display_factor, book.offers()[x].quantity());
+      }
+      // NaN out the remaining depth
+      for (size_t x = bids_size; x < COCONUT_MAX_DEPTH; ++x) {
+         cache.bids[x] = PriceSize::NO_MARKET;
+      }
+      for (size_t x = asks_size; x < COCONUT_MAX_DEPTH; ++x) {
+         cache.asks[x] = PriceSize::NO_MARKET;
+      }
+   }
+
    //
    // Trades
    void onTrade(onixs::Handler&, const onixs::Security&, const onixs::RecoveryEntryArgs&) override;
@@ -125,6 +178,7 @@ private:
    std::vector<std::string> _symbol_filter_storage;
    std::set<onixs::StrRef> _symbol_filter;
    std::vector<MDProducerChannel> _channels;
+   CMEOptions::BookType _booktype;
    std::atomic<int> _concurrency_monitor;
 
    struct BookData {

src/server/sparkd/md/platforms/mdp3_onixs/MDP3Main.cpp

@@ -213,7 +213,9 @@ unique_ptr<onixs::Handler> MDP3Main::prepare_handler(const string& channel_id) {
    settings.feeds().connectivityConfigurationFile(config_path());
 
    // Market data options
-   settings.bookManagement().consolidatedBooks().maintain(true);
+   if (options.book_type() == CMEOptions::BOOKTYPE_CONSOLIDATED) {
+      settings.bookManagement().consolidatedBooks().maintain(true);
+   }
 
    // 10k default is *far* too low
    settings.recoveryQueueMaxSize(options.recovery_queue_size());
@@ -334,12 +336,13 @@ unique_ptr<MDP3Listener> MDP3Main::prepare_listener(onixs::Handler& handler, con
       producer_channels.push_back(_producer.get_channel(symbol.symbol().c_str(), symbol.type(), symbol.exchange()));
   }
 
-   unique_ptr<MDP3Listener> listener(new MDP3Listener(_index, _producer, _security_recorder, _settlement_recorder, producer_channels));
+   const auto& options = channel_settings_for(channel.id);
+
+   unique_ptr<MDP3Listener> listener(new MDP3Listener(_index, _producer, _security_recorder, _settlement_recorder, producer_channels, options.book_type()));
    for (auto& symbol : channel.symbols) {
       listener->add_filter_symbol(symbol.symbol());
    }
 
-   const auto& options = channel_settings_for(channel.id);
    if (options.load_security_db()) {
       // Load the list of known securities and add them to the index for fast startup
       db::load_securities(channel.symbols, [&](const api::Security& security) {

src/server/sparkd/metrics/metrics.h

@@ -42,6 +42,7 @@ struct CounterMetric : public Metric {
    void decrement(T val) { _val -= val; }
    void set(T val) { _val = val; }
 };
+typedef CounterMetric<int> IntCounterMetric;
 
 template <class T>
 class AtomicCounterMetric : public Metric {

src/server/sparkd/orderman/actors/OrderStore.cpp

@@ -1,10 +1,10 @@
 #include "OrderStore.h"
 
-#include "lib/utils/PBUtils.h"
-#include "server/sparkd/persistence/SqliteStore.h"
 #include "lib/api/services/OrderService.pb.h"
-#include "server/sparkd/orderman/events/StoredOrderEvent.h"
+#include "lib/utils/PBUtils.h"
 #include "server/sparkd/orderman/events/OrderStateUpdateEvent.h"
+#include "server/sparkd/orderman/events/StoredOrderEvent.h"
+#include "server/sparkd/persistence/SqliteStore.h"
 
 #include <unordered_set>
 
@@ -13,74 +13,74 @@ namespace orderman {
 
 static auto const orders_ns = "orders";
 
-struct OrderStore::PImpl 
-{
-   PImpl() :
-      known_orders(),
-      store(new persistence::SqliteStore("orders")),
-      order_stream()
-   {}
+struct OrderStore::PImpl {
+   PImpl()
+       : known_orders(), store(new persistence::SqliteStore("orders")),
+         order_stream() {}
 
    const GatewayName gw_name;
-   std::unordered_map<OrderKey,std::unique_ptr<api::orderman::OrderRecord>> known_orders;
+   std::unordered_map<OrderKey, google::protobuf::Timestamp> known_orders;
    std::unique_ptr<persistence::IStore> store;
    mercury::Stream<StoredOrderEvent> order_stream;
 };
 
-OrderStore::OrderStore () : me (new PImpl()) {}
-OrderStore::~OrderStore () {}
+OrderStore::OrderStore() : me(new PImpl()) {}
+OrderStore::~OrderStore() {}
 
-void OrderStore::init()
-{
+void OrderStore::init() {
    api::orderman::OrderRecord rec;
    std::vector<StoredOrderEvent> stored_batch;
 
    auto all = me->store->retrieve_all()[orders_ns];
-   for (auto const &kvp : all)  
-      if (utils::pb::from_string<api::orderman::OrderRecord> (kvp.second,&rec)) {
+   for (auto const &kvp : all)
+      if (utils::pb::from_string<api::orderman::OrderRecord>(kvp.second,
+                                                             &rec)) {
          auto ok = OrderKey(std::stoi(kvp.first));
-         auto order_record = std::make_unique<api::orderman::OrderRecord> (rec);
-         stored_batch.emplace_back(*order_record.get());
-         me->known_orders[ok] = std::move(order_record);
-      }
-      else
-         LOG(OrderMan,Error,
+         stored_batch.emplace_back(rec);
+         me->known_orders[ok] = rec.state().created();
+      } else
+         LOG(OrderMan, Error,
              "Could not recover order {} from storage. "
-             "This should be investigated", kvp.first);
+             "This should be investigated",
+             kvp.first);
 
-   me->order_stream = make_stream<StoredOrderEvent> ();
-   LOG(OrderMan,Info,
-       "Order Store published {} orders recoved from disk", all.size());
-   publish_batch(me->order_stream,stored_batch);
-   subscribe<OrderStateUpdateEvent> ([this] (auto const&ev) {this->on_order_state_update(ev);});
+   me->order_stream = make_stream<StoredOrderEvent>();
+   LOG(OrderMan, Info, "Order Store published {} orders recoved from disk",
+       all.size());
+   publish_batch(me->order_stream, stored_batch);
+   subscribe<OrderStateUpdateEvent>(
+       [this](auto const &ev) { this->on_order_state_update(ev); });
 }
 
-void OrderStore::on_order_state_update(mercury::Update<OrderStateUpdateEvent> const&u)
-{
+void OrderStore::on_order_state_update(
+    mercury::Update<OrderStateUpdateEvent> const &u) {
    if (u.events)
       for (auto const &ev : *u.events) {
-         auto &ptr = me->known_orders[ev.key];
-         auto new_order = !ptr; 
-         if (new_order) {
-            ptr.reset (new api::orderman::OrderRecord());
-            *ptr->mutable_def() = *ev.definition;
-            ptr->set_key(ev.key.value());
-            ptr->set_parent_key(ev.parent_key.value());
-            *ptr->mutable_state()->mutable_created() = utils::pb::pb_now();
-         }
 
-         auto created = ptr->state().created(); 
-         *ptr->mutable_state() = ev.state;
-         *ptr->mutable_state()->mutable_created() = created;
-         *ptr->mutable_state()->mutable_last_change() = utils::pb::pb_now();
-         auto k = std::to_string(ptr->key());
-         auto v = utils::pb::to_string(*ptr);
-         if (new_order)
-            me->store->insert(orders_ns,k,v);
-         else
-            me->store->update(orders_ns,k,v);
-         publish(me->order_stream,*ptr.get());
+         api::orderman::OrderRecord order;
+         order.set_key(ev.key.value());
+         order.set_parent_key(ev.parent_key.value());
+         *order.mutable_def() = *ev.definition;
+         *order.mutable_state() = ev.state;
+         *order.mutable_state()->mutable_last_change() = utils::pb::pb_now();
+
+         auto it = me->known_orders.find(ev.key);
+         if (it == me->known_orders.end()) {
+            // new order
+            auto now = utils::pb::pb_now();
+            me->known_orders[ev.key] = now;
+            *order.mutable_state()->mutable_created() = now;
+            me->store->insert(orders_ns, std::to_string(order.key()),
+                              utils::pb::to_string(order));
+         } else {
+            *order.mutable_state()->mutable_created() = it->second;
+            me->store->update(orders_ns, std::to_string(order.key()),
+                              utils::pb::to_string(order));
+         }
+         LOG(OrderMan, Diag, "Storing order {}", utils::pb::to_string(order));
+         publish(me->order_stream, order);
       }
 }
 
-}}
+} // namespace orderman
+} // namespace bts

src/server/sparkd/orderman/actors/ServiceBridge.cpp

@@ -92,10 +92,10 @@ void ServiceBridge::on(mercury::Update<StoredOrderEvent> const &update) {
       bool was_ready = is_ready();
       me->have_orders = true;
 
-      std::vector<api::orderman::OrderRecord const *> orders;
+      std::vector<api::orderman::OrderRecord> orders;
       orders.reserve(update.events->size());
       for (auto ev : *(update.events)) {
-         me->known_orders.insert(OrderKey(ev.order_record->key()));
+         me->known_orders.insert(OrderKey(ev.order_record.key()));
          orders.push_back(ev.order_record);
       }
       me->on_orders(orders);

src/server/sparkd/orderman/actors/ServiceBridge.h

@@ -39,8 +39,8 @@ public:
    template <class EV> using OnEv = std::function<void(EV const&)>;
 
    using OnReady = std::function<void()>;
-   using OnOrders = std::function<void (std::vector<api::orderman::OrderRecord const*> )>;
-   using OnFills = std::function<void (std::vector<api::orderman::Fill const*> )>;
+   using OnOrders = std::function<void (std::vector<api::orderman::OrderRecord> const& )>;
+   using OnFills = std::function<void (std::vector<api::orderman::Fill const*> const&)>;
    
    ServiceBridge(OnReady const&, OnOrders const&, OnFills const&);
    void init() override;

src/server/sparkd/orderman/actors/order_manager/OrderManager.cpp

@@ -6,6 +6,7 @@
 #include "server/sparkd/orderman/actors/order_manager/OrderServicesData.h"
 #include "server/sparkd/orderman/events/NewAdminDataEvent.h"
 #include "server/sparkd/orderman/events/OrderManagerReadyEvent.h"
+#include "server/sparkd/orderman/events/StoredOrderEvent.h"
 #include "server/sparkd/orderman/order_factory/OrderFactory.h"
 #include "server/sparkd/orderman/orders/Order.h"
 #include "server/sparkd/securities/ISecurityDB.h"
@@ -96,8 +97,8 @@ void OrderManager::on_order_recovery(
          // recover orders from disk
          for (auto const &ev : *(update.events)) {
             auto &order =
-                me->services_data.order_factory.recover(*ev.order_record);
-            auto const &ostate = ev.order_record->state();
+                me->services_data.order_factory.recover(ev.order_record);
+            auto const &ostate = ev.order_record.state();
             OrderInfo oi{order, me->services_data};
             auto serv_ptr = oi.order_services.get();
             me->orders[order.key()] = std::move(oi);

src/server/sparkd/orderman/actors/order_manager/OrderServicesData.h

@@ -24,7 +24,6 @@
 #include "server/sparkd/orderman/events/NewSecuritiesEvent.h"
 #include "server/sparkd/orderman/events/OrderStateUpdateEvent.h"
 #include "server/sparkd/orderman/events/OutFIXMsgEvent.h"
-#include "server/sparkd/orderman/events/StoredOrderEvent.h"
 #include "server/sparkd/orderman/events/UpdateOrderEvent.h"
 #include "server/sparkd/orderman/limiter/Limiter.h"
 #include "server/sparkd/orderman/order_router/OrderRouter.h"

src/server/sparkd/orderman/events/StoredOrderEvent.h

@@ -5,6 +5,7 @@
 #include "lib/api/services/OrderService.pb.h"
 #include "server/sparkd/mercury/synopsis/MergeByKey.h"
 #include "lib/api/services/OrderService.pb.h"
+#include "server/sparkd/mercury/events/Event.h"
 
 namespace bts {
 namespace orderman {
@@ -14,10 +15,10 @@ class StoredOrderEvent : public mercury::Event <StoredOrderEvent>
 public:
    using SynopsisPolicy = mercury::MergeByKey<StoredOrderEvent>;
 
-   StoredOrderEvent(api::orderman::OrderRecord &record) : order_record(&record) {}
+   StoredOrderEvent(api::orderman::OrderRecord const &record) : order_record(record) {}
    
-   auto key() const {return order_record->key();}
-   api::orderman::OrderRecord *order_record;
+   auto key() const {return order_record.key();}
+   api::orderman::OrderRecord order_record;
 };
 
 }}

src/server/sparkd/orderman/orders/IcebergOrder.cpp

@@ -88,7 +88,8 @@ void IcebergOrder::on_init_recovered(api::orderman::OrderRecord const &rec,
    }
 }
 
-void IcebergOrder::on_activate(Price p, Quantity q, OrderServices &services, TickToTradeCorID) {
+void IcebergOrder::on_activate(Price p, Quantity q, OrderServices &services,
+                               TickToTradeCorID) {
    bts_assert(!me->services);
    me->services = &services;
    me->child_builder.activate(services);
@@ -184,6 +185,7 @@ void IcebergOrder::transition(api::orderman::OrderStatus new_status,
    auto const &to = OrderStatus_Name(new_status);
    LOG(OrderMan, Info, "Transitioning iceberg order {} from {} to {}. Msg: {}",
        key().value(), from, to, msg);
+   me->state.set_last_msg(msg);
    me->state.set_status(new_status);
 }
 
@@ -229,8 +231,9 @@ void IcebergOrder::send_next_child_order() {
    auto remaining = me->state.quantity() - me->state.filled_quantity();
    auto to_send = std::min(remaining, me->show_quantity.value());
    me->child.order = me->child_builder.build();
-   LOG(OrderMan, Info, "Iceberg order {} creating new child order {} with size "
-                       "{} price {}. Filled {} so far out of {}",
+   LOG(OrderMan, Info,
+       "Iceberg order {} creating new child order {} with size "
+       "{} price {}. Filled {} so far out of {}",
        key().value(), me->child.order.key().value(), to_send, me->state.price(),
        me->state.filled_quantity(), me->state.quantity());
 
@@ -251,14 +254,20 @@ void IcebergOrder::on_child_order_update(OrderStateUpdateEvent const &ev) {
             break;
 
          case ORDER_STATUS_COMPLETED: {
-            api::orderman::OrderState merged_state;
-            merge_states(me->state, me->child.state, &merged_state);
-            me->state = merged_state;
-            me->child.clear();
-            if (me->state.quantity() > me->state.filled_quantity())
-               send_next_child_order();
-            else
-               transition(ORDER_STATUS_COMPLETED);
+            if (me->child.state.filled_quantity() > 0) {
+               api::orderman::OrderState merged_state;
+               merge_states(me->state, me->child.state, &merged_state);
+               me->state = merged_state;
+               me->child.clear();
+               if (me->state.quantity() > me->state.filled_quantity())
+                  send_next_child_order();
+               else
+                  transition(ORDER_STATUS_COMPLETED);
+            } else {
+               me->child.clear();
+               transition(ORDER_STATUS_ERROR,
+                          "Child order completed without fills");
+            }
             break;
          }
 
@@ -302,5 +311,5 @@ void IcebergOrder::on_child_order_update(OrderStateUpdateEvent const &ev) {
    }
    report_state();
 }
-}
-}
+} // namespace orderman
+} // namespace bts

src/server/sparkd/orderman/orders/NativeOrder.cpp

@@ -126,9 +126,10 @@ void NativeOrder::on_init_recovered(OrderRecord const &rec,
 
       // bail if we can't find the security for the order
       if (!me->sec) {
-         LOG(OrderMan, Error, "Unknown security {} referenced by recovered "
-                              "order {}. This is a serious "
-                              "problem which has to be corrected manually. ",
+         LOG(OrderMan, Error,
+             "Unknown security {} referenced by recovered "
+             "order {}. This is a serious "
+             "problem which has to be corrected manually. ",
              def.security_key(), rec.key());
          transition_to_error(
              NativeOrderState::SUB_ERROR_UNKNOWN_SECURITY,
@@ -290,8 +291,9 @@ void NativeOrder::on_activate(Price p, Quantity q,
 
       // is our gateway connected?
       if (!me->session_services->is_gateway_connected()) {
-         LOG(OrderMan, Warn, "Attempt to send order to a disconnected gateway. "
-                             "Dropping order {}",
+         LOG(OrderMan, Warn,
+             "Attempt to send order to a disconnected gateway. "
+             "Dropping order {}",
              key().value());
          transition(ORDER_STATUS_REJECTED_BY_SPARK, "Exchange gateway is down");
          break;
@@ -302,7 +304,8 @@ void NativeOrder::on_activate(Price p, Quantity q,
           UserID(me->definition.user_id()),
           AccountID(me->definition.account_id()));
       if (!lqr.ok) {
-         LOG(OrderMan, Warn, "Unauthorized order attempted. User: {} on "
+         LOG(OrderMan, Warn,
+             "Unauthorized order attempted. User: {} on "
              "Account: {}. Msg: {} Order: {}. Dropping order",
              me->definition.user(), me->definition.account(), lqr.error_msg,
              key().value());
@@ -318,8 +321,9 @@ void NativeOrder::on_activate(Price p, Quantity q,
           me->definition.side(), q);
 
       if (!lqr.ok) {
-         LOG(OrderMan, Warn, "New order would breach limit. User: {} Account: "
-                             "{} Msg: {} Order: {}. Dropping order",
+         LOG(OrderMan, Warn,
+             "New order would breach limit. User: {} Account: "
+             "{} Msg: {} Order: {}. Dropping order",
              me->definition.user(), me->definition.account(), lqr.error_msg,
              key().value());
 
@@ -330,8 +334,9 @@ void NativeOrder::on_activate(Price p, Quantity q,
 
       // are we under the messaging threshold?
       if (!me->session_services->governor().can_send()) {
-         LOG(OrderMan, Warn, "New order would cause breach of messaging "
-                             "limits. Dropping order {}",
+         LOG(OrderMan, Warn,
+             "New order would cause breach of messaging "
+             "limits. Dropping order {}",
              key().value());
          transition(ORDER_STATUS_REJECTED_BY_SPARK,
                     "Messaging threshold breached");
@@ -747,8 +752,9 @@ void NativeOrder::report_state() {
              me->definition.side(),
              Quantity(previous_position.value() - current_position.value()));
    } else
-      LOG(OrderMan, Error, "Order {} ignored for purposes of limit enforcing "
-                           "because it references unknown security {} ",
+      LOG(OrderMan, Error,
+          "Order {} ignored for purposes of limit enforcing "
+          "because it references unknown security {} ",
           key().value(), me->definition.security_key());
 
    me->services->report_state(me->state);
@@ -845,8 +851,10 @@ void NativeOrder::init_fix() {
       auto stop_px = def.custom_def().stop().stop_price();
       auto encoded_stop_px =
           securities::encode_exchange_price(*me->sec, Price(stop_px)).value();
-      create_encoding.add<fix::StopPx>(encoded_stop_px, -1, -1);
-      update_encoding.add<fix::StopPx>(encoded_stop_px, -1, -1);
+      create_encoding.add<fix::StopPx>(encoded_stop_px,
+                                       me->price_encoding_precision, -1);
+      update_encoding.add<fix::StopPx>(encoded_stop_px,
+                                       me->price_encoding_precision, -1);
       break;
    }
 
@@ -860,8 +868,10 @@ void NativeOrder::init_fix() {
       auto stop_px = def.custom_def().stop_limit().stop_price();
       auto encoded_stop_px =
           securities::encode_exchange_price(*me->sec, Price(stop_px)).value();
-      create_encoding.add<fix::StopPx>(encoded_stop_px, -1, -1);
-      update_encoding.add<fix::StopPx>(encoded_stop_px, -1, -1);
+      create_encoding.add<fix::StopPx>(encoded_stop_px,
+                                       me->price_encoding_precision, -1);
+      update_encoding.add<fix::StopPx>(encoded_stop_px,
+                                       me->price_encoding_precision, -1);
 
       // price
       me->create_price_bm = create_encoding.add<fix::Price>(
@@ -1021,5 +1031,5 @@ NativeOrder::get_native_state(api::orderman::OrderState &st) {
    bts_assert(false);
    return *st.mutable_stop_limit()->mutable_native_state();
 }
-}
-}
+} // namespace orderman
+} // namespace bts

src/server/sparkd/orderman/orders/spreader_orders/spreader_states/SpreaderStartStateWorker.cpp

@@ -40,10 +40,13 @@ void SpreaderStartStateWorker::on_init_recovered(const bts::api::orderman::Order
                                                  const SpreaderPreActivationOrderServices& spreader_pao_services)
 {
    bts_assert(!_spreader->_inited());
-   // we are not handling recovery, so we go directly to the end state
-   // and the status of the order to cancelled
-   bts::api::orderman::OrderRecord order_record_copy = order_record;   
-   order_record_copy.mutable_state()->set_status(bts::api::orderman::OrderStatus::ORDER_STATUS_CANCELED);
+   // we are not handling recovery, so if order is not in a terminal state,
+   // instead of carrying on from where we left, we go directly to the end state
+   // and mark the order cancelled. if its already in a terminal state we 
+   // just go to end and not do anything
+   bts::api::orderman::OrderRecord order_record_copy = order_record;
+   if (!SpreaderUtils::order_in_terminal_state(order_record_copy.mutable_state()->status()))
+      order_record_copy.mutable_state()->set_status(bts::api::orderman::OrderStatus::ORDER_STATUS_CANCELED);
    _spreader->_bookkeeping_data_ptr = std::make_unique<SpreaderBookkeepingData>(*_spreader,
                                                                                 order_record_copy,
                                                                                 spreader_pao_services,

src/server/sparkd/orderman/server/OrderServiceServer.cpp

@@ -110,7 +110,7 @@ OrderServiceServer::~OrderServiceServer() {
 }
 
 void OrderServiceServer::on_orders(
-    std::vector<api::orderman::OrderRecord const *> const &orders) {
+    std::vector<api::orderman::OrderRecord> const &orders) {
    LOG(OrderMan, Info, "Received order update for publishing");
    std::vector<api::orderman::OrderRecord> created;
    created.reserve(orders.size());
@@ -118,12 +118,12 @@ void OrderServiceServer::on_orders(
    updated.reserve(orders.size());
 
    for (auto const &rec : orders) {
-      auto const &k = OrderKey(rec->key());
+      auto const &k = OrderKey(rec.key());
       if (me->known_orders.count(k) == 0) {
          me->known_orders.insert(k);
-         created.push_back(*rec);
+         created.push_back(rec);
       } else
-         updated.push_back(*rec);
+         updated.push_back(rec);
    }
 
    LOG(OrderMan, Info,

src/server/sparkd/orderman/server/OrderServiceServer.h

@@ -27,7 +27,7 @@ public:
 private:
    struct PImpl;
    std::unique_ptr<PImpl> me;
-   void on_orders(std::vector<api::orderman::OrderRecord const*> const&);
+   void on_orders(std::vector<api::orderman::OrderRecord> const&);
    void on_fills (std::vector<api::orderman::Fill const*> const&);
 };