htlcs: remove origin_htlc_id from htlc_out.

This is a transient field, so rework things so we don't leave it in
struct htlc_out.  Instead, load htlc_in first and connect htlc_out to
them as we go.

This also changes one place where we use it instead of the am_origin
flag.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

lightningd/htlc_end.c

@@ -28,6 +28,22 @@ struct htlc_in *find_htlc_in(const struct htlc_in_map *map,
 	return htlc_in_map_get(map, &key);
 }
 
+struct htlc_in *remove_htlc_in_by_dbid(struct htlc_in_map *remaining_htlcs_in,
+				       u64 dbid)
+{
+	struct htlc_in *hin;
+	struct htlc_in_map_iter ini;
+
+	for (hin = htlc_in_map_first(remaining_htlcs_in, &ini); hin;
+	     hin = htlc_in_map_next(remaining_htlcs_in, &ini)) {
+		if (hin->dbid == dbid) {
+			htlc_in_map_del(remaining_htlcs_in, hin);
+			return hin;
+		}
+	}
+	return NULL;
+}
+
 static void destroy_htlc_in(struct htlc_in *hend, struct htlc_in_map *map)
 {
 	htlc_in_map_del(map, hend);

lightningd/htlc_end.h

@@ -58,7 +58,6 @@ struct htlc_out {
 	 * is saved to the database, must be >0 after saving to the
 	 * database. */
 	u64 dbid;
-	u64 origin_htlc_id;
 	struct htlc_key key;
 	struct amount_msat msat;
 	u32 cltv_expiry;
@@ -123,6 +122,10 @@ struct htlc_in *find_htlc_in(const struct htlc_in_map *map,
 			     const struct channel *channel,
 			     u64 htlc_id);
 
+/* FIXME: Slow function only used at startup. */
+struct htlc_in *remove_htlc_in_by_dbid(struct htlc_in_map *remaining_htlcs_in,
+				       u64 dbid);
+
 struct htlc_out *find_htlc_out(const struct htlc_out_map *map,
 			       const struct channel *channel,
 			       u64 htlc_id);

lightningd/lightningd.c

@@ -630,7 +630,7 @@ int main(int argc, char *argv[])
 	int stop_fd;
 	struct timers *timers;
 	const char *stop_response;
-	struct htlc_in_map *unprocessed_htlcs;
+	struct htlc_in_map *unconnected_htlcs_in;
 	struct rlimit nofile = {1024, 1024};
 
 	/*~ Make sure that we limit ourselves to something reasonable. Modesty
@@ -778,7 +778,7 @@ int main(int argc, char *argv[])
 	 *  topology is initialized since some decisions rely on being able to
 	 *  know the blockheight. */
 	db_begin_transaction(ld->wallet->db);
-	unprocessed_htlcs = load_channels_from_wallet(ld);
+	unconnected_htlcs_in = load_channels_from_wallet(ld);
 	db_commit_transaction(ld->wallet->db);
 
 	/*~ Create RPC socket: now lightning-cli can send us JSON RPC commands
@@ -792,7 +792,7 @@ int main(int argc, char *argv[])
 	/*~ Process any HTLCs we were in the middle of when we exited, now
 	 * that plugins (who might want to know via htlc_accepted hook) are
 	 * active. */
-	htlcs_resubmit(ld, unprocessed_htlcs);
+	htlcs_resubmit(ld, unconnected_htlcs_in);
 
 	/*~ Activate connect daemon.  Needs to be after the initialization of
 	 * chaintopology, otherwise peers may connect and ask for

lightningd/peer_control.c

@@ -1579,27 +1579,47 @@ void activate_peers(struct lightningd *ld)
 struct htlc_in_map *load_channels_from_wallet(struct lightningd *ld)
 {
 	struct peer *peer;
+	struct htlc_in_map *unconnected_htlcs_in = tal(ld, struct htlc_in_map);
 
 	/* Load channels from database */
 	if (!wallet_init_channels(ld->wallet))
 		fatal("Could not load channels from the database");
 
-	/* This is a poor-man's db join :( */
+	/* First we load the incoming htlcs */
 	list_for_each(&ld->peers, peer, list) {
 		struct channel *channel;
 
 		list_for_each(&peer->channels, channel, list) {
-			if (!wallet_htlcs_load_for_channel(ld->wallet,
-							   channel,
-							   &ld->htlcs_in,
-							   &ld->htlcs_out)) {
+			if (!wallet_htlcs_load_in_for_channel(ld->wallet,
+							      channel,
+							      &ld->htlcs_in)) {
 				fatal("could not load htlcs for channel");
 			}
 		}
 	}
 
-	/* Now connect HTLC pointers together */
-	return htlcs_reconnect(ld, &ld->htlcs_in, &ld->htlcs_out);
+	/* Make a copy of the htlc_map: entries removed as they're matched */
+	htlc_in_map_copy(unconnected_htlcs_in, &ld->htlcs_in);
+
+	/* Now we load the outgoing HTLCs, so we can connect them. */
+	list_for_each(&ld->peers, peer, list) {
+		struct channel *channel;
+
+		list_for_each(&peer->channels, channel, list) {
+			if (!wallet_htlcs_load_out_for_channel(ld->wallet,
+							       channel,
+							       &ld->htlcs_out,
+							       unconnected_htlcs_in)) {
+				fatal("could not load outgoing htlcs for channel");
+			}
+		}
+	}
+
+#ifdef COMPAT_V061
+	fixup_htlcs_out(ld);
+#endif /* COMPAT_V061 */
+
+	return unconnected_htlcs_in;
 }
 
 static struct command_result *json_disconnect(struct command *cmd,

lightningd/peer_htlcs.c

@@ -1274,7 +1274,7 @@ static bool update_out_htlc(struct channel *channel,
 		}
 
 		/* For our own HTLCs, we commit payment to db lazily */
-		if (hout->origin_htlc_id == 0)
+		if (hout->am_origin)
 			payment_store(ld,
 				      &hout->payment_hash, hout->partid);
 	}
@@ -2089,94 +2089,35 @@ static void fixup_hout(struct lightningd *ld, struct htlc_out *hout)
 				  &hout->key.channel->peer->id),
 		   fix);
 }
-#endif /* COMPAT_V061 */
 
-/**
- * htlcs_reconnect -- Link outgoing HTLCs to their origins after initial db load
- *
- * For each outgoing HTLC find the incoming HTLC that triggered it. If
- * we are the origin of the transfer then we cannot resolve the
- * incoming HTLC in which case we just leave it `NULL`.
- *
- * Returns a map of any htlcs we need to retry.
- */
-struct htlc_in_map *htlcs_reconnect(struct lightningd *ld,
-				    struct htlc_in_map *htlcs_in,
-				    struct htlc_out_map *htlcs_out)
+void fixup_htlcs_out(struct lightningd *ld)
 {
-	struct htlc_in_map_iter ini;
 	struct htlc_out_map_iter outi;
-	struct htlc_in *hin;
 	struct htlc_out *hout;
-	struct htlc_in_map *unprocessed = tal(NULL, struct htlc_in_map);
-
-	/* Any HTLCs which happened to be incoming and weren't forwarded before
-	 * we shutdown/crashed: fail them now.
-	 *
-	 * Note that since we do local processing synchronously, so this never
-	 * captures local payments.  But if it did, it would be a tiny corner
-	 * case. */
-	htlc_in_map_init(unprocessed);
-	for (hin = htlc_in_map_first(htlcs_in, &ini); hin;
-	     hin = htlc_in_map_next(htlcs_in, &ini)) {
-		if (hin->hstate == RCVD_ADD_ACK_REVOCATION)
-			htlc_in_map_add(unprocessed, hin);
-	}
-
-	for (hout = htlc_out_map_first(htlcs_out, &outi); hout;
-	     hout = htlc_out_map_next(htlcs_out, &outi)) {
-
-		if (hout->am_origin) {
-			continue;
-		}
-
-		/* For fulfilled HTLCs, we fulfill incoming before outgoing is
-		 * completely resolved, so it's possible that we don't find
-		 * the incoming. */
-		for (hin = htlc_in_map_first(htlcs_in, &ini); hin;
-		     hin = htlc_in_map_next(htlcs_in, &ini)) {
-			if (hout->origin_htlc_id == hin->dbid) {
-				log_debug(ld->log,
-					  "Found corresponding htlc_in %" PRIu64
-					  " for htlc_out %" PRIu64,
-					  hin->dbid, hout->dbid);
-				htlc_out_connect_htlc_in(hout, hin);
-				break;
-			}
-		}
-
-		if (!hout->in && !hout->preimage) {
-#ifdef COMPAT_V061
-			log_broken(ld->log,
-				   "Missing preimage for orphaned HTLC; replacing with zeros");
-			hout->preimage = talz(hout, struct preimage);
-#else
-			fatal("Unable to find corresponding htlc_in %"PRIu64
-			      " for unfulfilled htlc_out %"PRIu64,
-			      hout->origin_htlc_id, hout->dbid);
-#endif
-		}
-#ifdef COMPAT_V061
-		fixup_hout(ld, hout);
-#endif
 
-		if (hout->in)
-			htlc_in_map_del(unprocessed, hout->in);
+	for (hout = htlc_out_map_first(&ld->htlcs_out, &outi);
+	     hout;
+	     hout = htlc_out_map_next(&ld->htlcs_out, &outi)) {
+		if (!hout->am_origin)
+			fixup_hout(ld, hout);
 	}
-
-	return unprocessed;
 }
+#endif /* COMPAT_V061 */
 
-void htlcs_resubmit(struct lightningd *ld, struct htlc_in_map *unprocessed)
+void htlcs_resubmit(struct lightningd *ld,
+		    struct htlc_in_map *unconnected_htlcs_in)
 {
 	struct htlc_in *hin;
 	struct htlc_in_map_iter ini;
 	enum onion_type failcode COMPILER_WANTS_INIT("gcc7.4.0 bad, 8.3 OK");
 
-	/* Now fail any which were stuck. */
-	for (hin = htlc_in_map_first(unprocessed, &ini);
+	/* Now retry any which were stuck. */
+	for (hin = htlc_in_map_first(unconnected_htlcs_in, &ini);
 	     hin;
-	     hin = htlc_in_map_next(unprocessed, &ini)) {
+	     hin = htlc_in_map_next(unconnected_htlcs_in, &ini)) {
+		if (hin->hstate != RCVD_ADD_ACK_REVOCATION)
+			continue;
+
 		log_unusual(hin->key.channel->log,
 			    "Replaying old unprocessed HTLC #%"PRIu64,
 			    hin->key.id);
@@ -2192,8 +2133,8 @@ void htlcs_resubmit(struct lightningd *ld, struct htlc_in_map *unprocessed)
 	}
 
 	/* Don't leak memory! */
-	htlc_in_map_clear(unprocessed);
-	tal_free(unprocessed);
+	htlc_in_map_clear(unconnected_htlcs_in);
+	tal_free(unconnected_htlcs_in);
 }
 
 #if DEVELOPER

lightningd/peer_htlcs.h

@@ -64,11 +64,11 @@ void onchain_fulfilled_htlc(struct channel *channel,
 
 void htlcs_notify_new_block(struct lightningd *ld, u32 height);
 
-struct htlc_in_map *htlcs_reconnect(struct lightningd *ld,
-				    struct htlc_in_map *htlcs_in,
-				    struct htlc_out_map *htlcs_out);
+/* Only defined if COMPAT_V061 */
+void fixup_htlcs_out(struct lightningd *ld);
 
-void htlcs_resubmit(struct lightningd *ld, struct htlc_in_map *unprocessed);
+void htlcs_resubmit(struct lightningd *ld,
+		    struct htlc_in_map *unconnected_htlcs_in);
 
 /* For HTLCs which terminate here, invoice payment calls one of these. */
 void fulfill_htlc(struct htlc_in *hin, const struct preimage *preimage);

lightningd/test/run-find_my_abspath.c

@@ -97,7 +97,8 @@ void hsm_init(struct lightningd *ld UNNEEDED)
 void htlcs_notify_new_block(struct lightningd *ld UNNEEDED, u32 height UNNEEDED)
 { fprintf(stderr, "htlcs_notify_new_block called!\n"); abort(); }
 /* Generated stub for htlcs_resubmit */
-void htlcs_resubmit(struct lightningd *ld UNNEEDED, struct htlc_in_map *unprocessed UNNEEDED)
+void htlcs_resubmit(struct lightningd *ld UNNEEDED,
+		    struct htlc_in_map *unconnected_htlcs_in UNNEEDED)
 { fprintf(stderr, "htlcs_resubmit called!\n"); abort(); }
 /* Generated stub for jsonrpc_listen */
 void jsonrpc_listen(struct jsonrpc *rpc UNNEEDED, struct lightningd *ld UNNEEDED)

lightningd/test/run-invoice-select-inchan.c

@@ -96,6 +96,9 @@ void   fatal(const char *fmt UNNEEDED, ...)
 /* Generated stub for feature_is_set */
 bool feature_is_set(const u8 *features UNNEEDED, size_t bit UNNEEDED)
 { fprintf(stderr, "feature_is_set called!\n"); abort(); }
+/* Generated stub for fixup_htlcs_out */
+void fixup_htlcs_out(struct lightningd *ld UNNEEDED)
+{ fprintf(stderr, "fixup_htlcs_out called!\n"); abort(); }
 /* Generated stub for fromwire_channel_dev_memleak_reply */
 bool fromwire_channel_dev_memleak_reply(const void *p UNNEEDED, bool *leak UNNEEDED)
 { fprintf(stderr, "fromwire_channel_dev_memleak_reply called!\n"); abort(); }
@@ -133,11 +136,6 @@ bool htlc_is_trimmed(enum side htlc_owner UNNEEDED,
 		     struct amount_sat dust_limit UNNEEDED,
 		     enum side side UNNEEDED)
 { fprintf(stderr, "htlc_is_trimmed called!\n"); abort(); }
-/* Generated stub for htlcs_reconnect */
-struct htlc_in_map *htlcs_reconnect(struct lightningd *ld UNNEEDED,
-				    struct htlc_in_map *htlcs_in UNNEEDED,
-				    struct htlc_out_map *htlcs_out UNNEEDED)
-{ fprintf(stderr, "htlcs_reconnect called!\n"); abort(); }
 /* Generated stub for json_add_address */
 void json_add_address(struct json_stream *response UNNEEDED, const char *fieldname UNNEEDED,
 		      const struct wireaddr *addr UNNEEDED)
@@ -504,12 +502,17 @@ void wallet_channeltxs_add(struct wallet *w UNNEEDED, struct channel *chan UNNEE
 			    const int type UNNEEDED, const struct bitcoin_txid *txid UNNEEDED,
 			   const u32 input_num UNNEEDED, const u32 blockheight UNNEEDED)
 { fprintf(stderr, "wallet_channeltxs_add called!\n"); abort(); }
-/* Generated stub for wallet_htlcs_load_for_channel */
-bool wallet_htlcs_load_for_channel(struct wallet *wallet UNNEEDED,
-				   struct channel *chan UNNEEDED,
-				   struct htlc_in_map *htlcs_in UNNEEDED,
-				   struct htlc_out_map *htlcs_out UNNEEDED)
-{ fprintf(stderr, "wallet_htlcs_load_for_channel called!\n"); abort(); }
+/* Generated stub for wallet_htlcs_load_in_for_channel */
+bool wallet_htlcs_load_in_for_channel(struct wallet *wallet UNNEEDED,
+				      struct channel *chan UNNEEDED,
+				      struct htlc_in_map *htlcs_in UNNEEDED)
+{ fprintf(stderr, "wallet_htlcs_load_in_for_channel called!\n"); abort(); }
+/* Generated stub for wallet_htlcs_load_out_for_channel */
+bool wallet_htlcs_load_out_for_channel(struct wallet *wallet UNNEEDED,
+				       struct channel *chan UNNEEDED,
+				       struct htlc_out_map *htlcs_out UNNEEDED,
+				       struct htlc_in_map *remaining_htlcs_in UNNEEDED)
+{ fprintf(stderr, "wallet_htlcs_load_out_for_channel called!\n"); abort(); }
 /* Generated stub for wallet_init_channels */
 bool wallet_init_channels(struct wallet *w UNNEEDED)
 { fprintf(stderr, "wallet_init_channels called!\n"); abort(); }

wallet/test/run-wallet.c

@@ -1156,7 +1156,7 @@ static bool test_htlc_crud(struct lightningd *ld, const tal_t *ctx)
 	struct channel *chan = tal(ctx, struct channel);
 	struct peer *peer = talz(ctx, struct peer);
 	struct wallet *w = create_test_wallet(ld, ctx);
-	struct htlc_in_map *htlcs_in = tal(ctx, struct htlc_in_map);
+	struct htlc_in_map *htlcs_in = tal(ctx, struct htlc_in_map), *rem;
 	struct htlc_out_map *htlcs_out = tal(ctx, struct htlc_out_map);
 
 	/* Make sure we have our references correct */
@@ -1220,11 +1220,16 @@ static bool test_htlc_crud(struct lightningd *ld, const tal_t *ctx)
 	db_begin_transaction(w->db);
 	CHECK(!wallet_err);
 
-	CHECK_MSG(wallet_htlcs_load_for_channel(w, chan, htlcs_in, htlcs_out),
-		  "Failed loading HTLCs");
+	CHECK_MSG(wallet_htlcs_load_in_for_channel(w, chan, htlcs_in),
+		  "Failed loading in HTLCs");
+	/* Freed by htlcs_resubmit */
+	rem = tal(NULL, struct htlc_in_map);
+	htlc_in_map_copy(rem, htlcs_in);
+	CHECK_MSG(wallet_htlcs_load_out_for_channel(w, chan, htlcs_out, rem),
+		  "Failed loading out HTLCs");
 	db_commit_transaction(w->db);
 
-	htlcs_resubmit(w->ld, htlcs_reconnect(w->ld, htlcs_in, htlcs_out));
+	htlcs_resubmit(w->ld, rem);
 	CHECK(!wallet_err);
 
 	hin = htlc_in_map_get(htlcs_in, &in.key);