PeerComplaintsDesk: read peer_connected and persist deferred-close times

Fixes two issues from review of the auto-close patience change:

- The connectivity check read "connected", but listpeerchannels
  reports the field as "peer_connected".  Every peer was treated
  as offline, so the mutual-close path never fired and every close
  would eventually take the unilateral path.  Field name verified
  against a live node.

- close_pending was in-memory only, so a restart reset the
  patience window; routine redeploys could postpone the unilateral
  fallback indefinitely.  Persist the first-deferred time in a new
  PeerComplaintsDesk_closepending table, restore it at startup,
  and clear it when the peer comes back online, drops below the
  close threshold, or the channel is destroyed.

Boss/Mod/PeerComplaintsDesk/Main.cpp

@@ -124,8 +124,9 @@ private:
 		bus.subscribe<Msg::DbResource
 			     >([this](Msg::DbResource const& m) {
 			db = m.db;
-			return db.transact().then([](Sqlite3::Tx tx) {
+			return db.transact().then([this](Sqlite3::Tx tx) {
 				Recorder::initialize(tx);
+				close_pending = Recorder::get_close_pendings(tx);
 				tx.commit();
 
 				return Ev::lift();
@@ -274,7 +275,6 @@ private:
 	Ev::Io<void> check_close() {
 		return db.transact().then([this](Sqlite3::Tx tx) {
 			auto complaints = Recorder::check_complaints(tx);
-			tx.commit();
 
 			auto const& unmanaged = unmanager.get_unmanaged();
 
@@ -320,11 +320,16 @@ private:
 				if (std::find( to_close.begin()
 					     , to_close.end()
 					     , it->first
-					     ) == to_close.end())
+					     ) == to_close.end()) {
+					Recorder::clear_close_pending( tx
+								     , it->first
+								     );
 					it = close_pending.erase(it);
-				else
+				} else
 					++it;
 			}
+			tx.commit();
+
 			if (!first)
 				act += Boss::log( bus, Debug
 						, "PeerComplaintsDesk: Complaints: %s"
@@ -373,9 +378,9 @@ private:
 			if (res.has("channels")) {
 				auto cs = res["channels"];
 				for (auto c : cs) {
-					if (!c.has("connected"))
+					if (!c.has("peer_connected"))
 						continue;
-					auto conn = c["connected"];
+					auto conn = c["peer_connected"];
 					if (!conn.is_boolean())
 						continue;
 					if (bool(conn)) {
@@ -393,21 +398,39 @@ private:
 				 * peers are rare.
 				 */
 				close_pending.erase(p);
-				return do_close(p);
+				return db.transact(
+				).then([this, p](Sqlite3::Tx tx) {
+					Recorder::clear_close_pending(tx, p);
+					tx.commit();
+					return Ev::lift();
+				}).then([this, p]() {
+					return do_close(p);
+				});
 			}
 			/* Offline: wait for a mutual-close window
 			 * until patience runs out.
 			 */
 			auto it = close_pending.find(p);
 			if (it == close_pending.end()) {
-				close_pending[p] = Ev::now();
-				return Boss::log( bus, Info
-						, "PeerComplaintsDesk: %s is "
-						  "not connected, deferring "
-						  "close while waiting for a "
-						  "mutual-close window."
-						, Util::stringify(p).c_str()
-						);
+				auto since = Ev::now();
+				close_pending[p] = since;
+				/* Persist so a restart does not reset
+				 * the patience window.
+				 */
+				return db.transact(
+				).then([this, p, since](Sqlite3::Tx tx) {
+					Recorder::note_close_pending(tx, p, since);
+					tx.commit();
+					return Ev::lift();
+				}).then([this, p]() {
+					return Boss::log( bus, Info
+							, "PeerComplaintsDesk: %s is "
+							  "not connected, deferring "
+							  "close while waiting for a "
+							  "mutual-close window."
+							, Util::stringify(p).c_str()
+							);
+				});
 			}
 			if (Ev::now() - it->second < close_patience)
 				return Boss::log( bus, Debug
@@ -458,7 +481,9 @@ private:
 		});
 	}
 	Ev::Io<void> on_channel_destroy(Ln::NodeId const& p) {
-		return db.transact().then([p](Sqlite3::Tx tx) {
+		return db.transact().then([this, p](Sqlite3::Tx tx) {
+			close_pending.erase(p);
+			Recorder::clear_close_pending(tx, p);
 			Recorder::channel_closed(tx, p);
 			tx.commit();
 			return Ev::lift();

Boss/Mod/PeerComplaintsDesk/Recorder.cpp

@@ -53,6 +53,15 @@ void Recorder::initialize(Sqlite3::Tx& tx) {
 	       "PeerComplaintsDesk_closedcomplaints_time"
 	    ON "PeerComplaintsDesk_closedcomplaints"(closedtime)
 	     ;
+
+	CREATE TABLE IF NOT EXISTS
+	       "PeerComplaintsDesk_closepending"
+	     ( peerdbid INTEGER PRIMARY KEY
+	     , since REAL NOT NULL
+	     , FOREIGN KEY(peerdbid)
+	       REFERENCES "PeerComplaintsDesk_peers"(peerdbid)
+	       ON DELETE CASCADE
+	     );
 	)QRY");
 }
 void Recorder::cleanup( Sqlite3::Tx& tx
@@ -286,6 +295,52 @@ void Recorder::channel_closed(Sqlite3::Tx& tx, Ln::NodeId const& nid) {
 		.execute()
 		;
 }
+void Recorder::note_close_pending( Sqlite3::Tx& tx
+				 , Ln::NodeId const& peer
+				 , double since
+				 ) {
+	auto peerdbid = get_peerdbid(tx, peer);
+	tx.query(R"QRY(
+	INSERT OR IGNORE INTO "PeerComplaintsDesk_closepending"
+	      (  peerdbid,  since)
+	VALUES( :peerdbid, :since);
+	)QRY")
+		.bind(":peerdbid", peerdbid)
+		.bind(":since", since)
+		.execute()
+		;
+}
+std::map< Ln::NodeId
+	, double
+	> Recorder::get_close_pendings(Sqlite3::Tx& tx) {
+	auto rv = std::map<Ln::NodeId, double>();
+
+	auto fetch = tx.query(R"QRY(
+	SELECT nodeid, since
+	  FROM "PeerComplaintsDesk_peers" NATURAL JOIN
+	       "PeerComplaintsDesk_closepending"
+	    ;
+	)QRY").execute();
+	for (auto& r : fetch)
+		rv[Ln::NodeId(r.get<std::string>(0))] = r.get<double>(1);
+
+	return rv;
+}
+void Recorder::clear_close_pending( Sqlite3::Tx& tx
+				  , Ln::NodeId const& peer
+				  ) {
+	tx.query(R"QRY(
+	DELETE FROM "PeerComplaintsDesk_closepending"
+	 WHERE peerdbid = (SELECT peerdbid
+	                     FROM "PeerComplaintsDesk_peers"
+	                    WHERE nodeid = :nodeid)
+	     ;
+	)QRY")
+		.bind(":nodeid", std::string(peer))
+		.execute()
+		;
+}
+
 std::map< Ln::NodeId
 	, std::vector<std::string>
 	> Recorder::get_closed_complaints(Sqlite3::Tx& tx) {

Boss/Mod/PeerComplaintsDesk/Recorder.hpp

@@ -81,6 +81,30 @@ namespace Recorder {
 	 * separate area.
 	 */
 	void channel_closed(Sqlite3::Tx&, Ln::NodeId const&);
+
+	/** Boss::Mod::PeerComplaintsDesk::Recorder::note_close_pending
+	 *
+	 * @brief records the time a close was first deferred
+	 * because the peer was offline.  Does not change an
+	 * existing record.
+	 */
+	void note_close_pending( Sqlite3::Tx&
+			       , Ln::NodeId const&
+			       , double since
+			       );
+	/** Boss::Mod::PeerComplaintsDesk::Recorder::get_close_pendings
+	 *
+	 * @brief gathers the first-deferred times of all peers
+	 * with a deferred close.
+	 */
+	std::map< Ln::NodeId
+		, double
+		> get_close_pendings(Sqlite3::Tx&);
+	/** Boss::Mod::PeerComplaintsDesk::Recorder::clear_close_pending
+	 *
+	 * @brief forgets the deferred-close time of a peer.
+	 */
+	void clear_close_pending(Sqlite3::Tx&, Ln::NodeId const&);
 	/** Boss::Mod::PeerComplaintsDesk::Recorder::get_closed_complaints
 	 *
 	 * @brief gathers all remembered non-ignored complaints for