bwatch: notify watchman on block_processed

diff --git a/plugins/bwatch/bwatch.c b/plugins/bwatch/bwatch.c
index 097375ed..d94328a7 100644
--- a/plugins/bwatch/bwatch.c
+++ b/plugins/bwatch/bwatch.c
@@ -79,20 +79,9 @@ static struct command_result *poll_finished(struct command *cmd)
 	return timer_complete(cmd);
 }
 
-/* Just persisted a block — there may be more to catch up to, so poll again
- * immediately rather than waiting for the full interval.  Once getchaininfo
- * reports no change, poll_finished resets us to the steady-state cadence. */
-static struct command_result *fetch_more(struct command *cmd)
-{
-	struct bwatch *bwatch = bwatch_of(cmd->plugin);
-
-	bwatch->poll_timer = global_timer(cmd->plugin, time_from_sec(0),
-					  bwatch_poll_chain, NULL);
-	return timer_complete(cmd);
-}
-
 /* Process one block fetched from bitcoind: update tip, append to history,
- * then persist; the poll is rescheduled once the datastore write completes. */
+ * then persist; once persisted we notify watchman, and the next poll is
+ * scheduled from the block_processed ack so we don't race ahead of it. */
 static struct command_result *handle_block(struct command *cmd,
 					   const char *method UNUSED,
 					   const char *buf,
@@ -123,7 +112,8 @@ static struct command_result *handle_block(struct command *cmd,
 		bwatch->current_blockhash,
 		block->hdr.prev_hash,
 	};
-	return bwatch_add_block_to_datastore(cmd, &br, fetch_more);
+	return bwatch_add_block_to_datastore(cmd, &br,
+					     bwatch_send_block_processed);
 }
 
 /* getchaininfo response: pick the next block to fetch (or just reschedule). */
diff --git a/plugins/bwatch/bwatch_interface.c b/plugins/bwatch/bwatch_interface.c
index aba4a221..6b2893ba 100644
--- a/plugins/bwatch/bwatch_interface.c
+++ b/plugins/bwatch/bwatch_interface.c
@@ -1,2 +1,77 @@
 #include "config.h"
+#include <common/json_param.h>
+#include <common/json_parse.h>
+#include <common/json_stream.h>
 #include <plugins/bwatch/bwatch_interface.h>
+
+/*
+ * ============================================================================
+ * SENDING BLOCK_PROCESSED NOTIFICATION
+ *
+ * After bwatch has persisted a new tip, it tells watchman by sending the
+ * block_processed RPC.  The next poll is scheduled from the ack callback,
+ * which guarantees watchman's persisted height is updated before bwatch
+ * looks for another block — important for crash safety: on restart we
+ * trust watchman's height as the floor and re-fetch anything above it.
+ * ============================================================================
+ */
+
+/* Watchman acked block_processed: safe to poll for the next block. */
+static struct command_result *block_processed_ack(struct command *cmd,
+						  const char *method UNUSED,
+						  const char *buf,
+						  const jsmntok_t *result,
+						  void *unused UNUSED)
+{
+	struct bwatch *bwatch = bwatch_of(cmd->plugin);
+	u32 acked_height;
+	const char *err;
+
+	err = json_scan(tmpctx, buf, result,
+			"{blockheight:%}",
+			JSON_SCAN(json_to_number, &acked_height));
+	if (err)
+		plugin_err(cmd->plugin, "block_processed ack '%.*s': %s",
+			   json_tok_full_len(result),
+			   json_tok_full(buf, result), err);
+
+	plugin_log(cmd->plugin, LOG_DBG,
+		   "Received block_processed ack for height %u", acked_height);
+
+	bwatch->poll_timer = global_timer(cmd->plugin, time_from_sec(0),
+					  bwatch_poll_chain, NULL);
+	return timer_complete(cmd);
+}
+
+/* Non-fatal: watchman may not be ready yet (e.g. lightningd still booting).
+ * Reschedule the poll anyway so we keep retrying without busy-looping. */
+static struct command_result *block_processed_err(struct command *cmd,
+						  const char *method UNUSED,
+						  const char *buf,
+						  const jsmntok_t *result,
+						  void *unused UNUSED)
+{
+	struct bwatch *bwatch = bwatch_of(cmd->plugin);
+
+	plugin_log(cmd->plugin, LOG_BROKEN,
+		   "block_processed RPC failed (watchman not ready?): %.*s",
+		   json_tok_full_len(result), json_tok_full(buf, result));
+
+	bwatch->poll_timer = global_timer(cmd->plugin, time_from_sec(0),
+					  bwatch_poll_chain, NULL);
+	return timer_complete(cmd);
+}
+
+struct command_result *bwatch_send_block_processed(struct command *cmd)
+{
+	struct bwatch *bwatch = bwatch_of(cmd->plugin);
+	struct out_req *req;
+
+	req = jsonrpc_request_start(cmd, "block_processed",
+				    block_processed_ack, block_processed_err,
+				    NULL);
+	json_add_u32(req->js, "blockheight", bwatch->current_height);
+	json_add_string(req->js, "blockhash",
+			fmt_bitcoin_blkid(tmpctx, &bwatch->current_blockhash));
+	return send_outreq(req);
+}
diff --git a/plugins/bwatch/bwatch_interface.h b/plugins/bwatch/bwatch_interface.h
index 944a66f0..e3424885 100644
--- a/plugins/bwatch/bwatch_interface.h
+++ b/plugins/bwatch/bwatch_interface.h
@@ -6,7 +6,13 @@
 
 /* Outward-facing interface from bwatch to lightningd.
  *
- * Subsequent commits add the watch_found / watch_revert / block_processed
- * notifications and the addwatch / delwatch / listwatch RPC commands. */
+ * Subsequent commits add the watch_found / watch_revert notifications
+ * and the addwatch / delwatch / listwatch RPC commands. */
+
+/* Send a block_processed RPC to watchman after a new block has been
+ * persisted.  The next poll is started from the ack callback so we don't
+ * race ahead of watchman's view of the chain.  Chains on the same poll
+ * command so timer_complete fires once watchman has acknowledged. */
+struct command_result *bwatch_send_block_processed(struct command *cmd);
 
 #endif /* LIGHTNING_PLUGINS_BWATCH_BWATCH_INTERFACE_H */