diff --git a/frontend/mempool.go b/frontend/mempool.go new file mode 100644 index 0000000..0d8d302 --- /dev/null +++ b/frontend/mempool.go @@ -0,0 +1,173 @@ +package frontend + +import ( + "encoding/hex" + "encoding/json" + "sync" + "time" + + "github.com/btcsuite/btcd/rpcclient" + "github.com/sirupsen/logrus" + + "git.hush.is/hush/lightwalletd/common" + "git.hush.is/hush/lightwalletd/walletrpc" +) + +// mempoolMonitor polls the node's mempool ONCE (regardless of how many wallets are streaming) and +// fans each new transaction out to every subscribed GetMempoolStream handler. This replaces +// per-connection polling: N connected wallets no longer each hit getrawmempool/getrawtransaction. +// +// Lifecycle matches the semantics the client's monitor loop expects: while a block is current the +// monitor emits each mempool tx to subscribers exactly once; when a new block is mined it resets +// and closes all subscriber channels, so each handler returns (closing its stream) and the client +// reconnects after re-syncing the block. The poller starts lazily on the first subscription and, +// when no clients are connected, idles without touching the node. +type mempoolMonitor struct { + mu sync.Mutex + subs map[int]chan *walletrpc.RawTransaction + nextID int + seen map[string]bool // txids already emitted for the current block + ordered []*walletrpc.RawTransaction // emitted txs in arrival order, replayed to late subscribers +} + +var ( + sharedMempoolMonitor *mempoolMonitor + mempoolMonitorOnce sync.Once +) + +// getMempoolMonitor returns the process-wide mempool monitor, starting its poller on first use. +// The client/cache/log are bound once (they are process singletons on the SqlStreamer). +func getMempoolMonitor(client *rpcclient.Client, cache *common.BlockCache, log *logrus.Entry) *mempoolMonitor { + mempoolMonitorOnce.Do(func() { + sharedMempoolMonitor = &mempoolMonitor{ + subs: make(map[int]chan *walletrpc.RawTransaction), + seen: make(map[string]bool), + } + go sharedMempoolMonitor.run(client, cache, log) + }) + return sharedMempoolMonitor +} + +// subscribe registers a stream. It returns the subscriber id, a channel of subsequent mempool txs, +// and a snapshot of txs already emitted this block (to be sent first). Registration and snapshot +// are taken atomically, so every tx reaches a subscriber exactly once (snapshot xor channel). +func (m *mempoolMonitor) subscribe() (int, chan *walletrpc.RawTransaction, []*walletrpc.RawTransaction) { + m.mu.Lock() + defer m.mu.Unlock() + id := m.nextID + m.nextID++ + ch := make(chan *walletrpc.RawTransaction, 256) + m.subs[id] = ch + snapshot := make([]*walletrpc.RawTransaction, len(m.ordered)) + copy(snapshot, m.ordered) + return id, ch, snapshot +} + +// unsubscribe removes a stream. Safe to call after a new-block reset already dropped it (the map +// lookup guards against a double close). +func (m *mempoolMonitor) unsubscribe(id int) { + m.mu.Lock() + defer m.mu.Unlock() + if ch, ok := m.subs[id]; ok { + delete(m.subs, id) + close(ch) + } +} + +// reset is called when a new block is mined: forget this block's txs and close every subscriber +// stream so clients re-sync the block and reconnect. +func (m *mempoolMonitor) reset() { + m.mu.Lock() + defer m.mu.Unlock() + m.seen = make(map[string]bool) + m.ordered = nil + for id, ch := range m.subs { + delete(m.subs, id) + close(ch) + } +} + +func (m *mempoolMonitor) run(client *rpcclient.Client, cache *common.BlockCache, log *logrus.Entry) { + lastHeight := cache.GetLatestBlock() + for { + // Track block height even while idle, so a new subscriber isn't immediately closed by a + // height that advanced during the idle period. + if h := cache.GetLatestBlock(); h != lastHeight { + lastHeight = h + m.reset() + } + + m.mu.Lock() + idle := len(m.subs) == 0 + m.mu.Unlock() + if idle { + // No one is listening; don't touch the node until a client subscribes. + time.Sleep(2 * time.Second) + continue + } + + height := uint64(lastHeight + 1) + + // List the current mempool txids: getrawmempool(false) -> ["txid", ...]. + mpParams := []json.RawMessage{json.RawMessage("false")} + result, rpcErr := client.RawRequest("getrawmempool", mpParams) + if rpcErr != nil { + log.Warningf("mempool monitor: getrawmempool failed: %s", rpcErr.Error()) + time.Sleep(2 * time.Second) + continue + } + var txids []string + if err := json.Unmarshal(result, &txids); err != nil { + log.Warningf("mempool monitor: cannot parse getrawmempool: %s", err.Error()) + time.Sleep(2 * time.Second) + continue + } + + for _, txid := range txids { + m.mu.Lock() + already := m.seen[txid] + m.mu.Unlock() + if already { + continue + } + + // Fetch the full serialized tx: getrawtransaction("txid") -> hex. Full bytes are + // required so shielded memos survive (a CompactTx's 52-byte prefix would not carry a + // memo). The mempool txid is already in RPC display byte order, so (unlike + // GetTransaction) no reversal is needed. + txParams := []json.RawMessage{json.RawMessage("\"" + txid + "\"")} + txResult, txErr := client.RawRequest("getrawtransaction", txParams) + if txErr != nil { + // The tx may have been mined or evicted between listing and fetch; skip it. + continue + } + var txhex string + if err := json.Unmarshal(txResult, &txhex); err != nil { + continue + } + txBytes, err := hex.DecodeString(txhex) + if err != nil { + continue + } + rtx := &walletrpc.RawTransaction{Data: txBytes, Height: height} + + // Record + fan out under the lock. The send is non-blocking (default case), so a slow + // subscriber never stalls the poller or the other subscribers. + m.mu.Lock() + if !m.seen[txid] { + m.seen[txid] = true + m.ordered = append(m.ordered, rtx) + for _, ch := range m.subs { + select { + case ch <- rtx: + default: + // Slow subscriber; drop. Block sync will deliver the confirmed tx later. + } + } + } + m.mu.Unlock() + } + + time.Sleep(2 * time.Second) + } +} diff --git a/frontend/service.go b/frontend/service.go index 14ece28..f8b66fc 100644 --- a/frontend/service.go +++ b/frontend/service.go @@ -116,6 +116,45 @@ func (s *SqlStreamer) GetAddressTxids(addressBlockFilter *walletrpc.TransparentA return nil } +// GetMempoolStream streams currently-unconfirmed (mempool) transactions to the client as full +// RawTransactions, so lightwallets can detect 0-confirmation transactions -- including reading +// shielded memos (e.g. incoming chat messages) -- without waiting for a block. Each mempool tx is +// emitted once; the stream stays open until a new block is mined, at which point it returns +// (closing the stream) so the client re-syncs the block and reconnects. This mirrors the +// zecwallet/Hush lightwalletd semantics the wallets already expect. +// +// The node's mempool is polled by a single process-wide monitor that fans out to all subscribers +// (see mempool.go), so the node isn't polled once per connected wallet. +func (s *SqlStreamer) GetMempoolStream(_ *walletrpc.Empty, resp walletrpc.CompactTxStreamer_GetMempoolStreamServer) error { + monitor := getMempoolMonitor(s.client, s.cache, s.log) + id, ch, snapshot := monitor.subscribe() + defer monitor.unsubscribe(id) + + // First send the txs already in this block's mempool, so a wallet that connects mid-block still + // sees them. + for _, rtx := range snapshot { + if err := resp.Send(rtx); err != nil { + return err + } + } + + // Then stream subsequent txs until the monitor closes the channel (a new block was mined) or + // the client disconnects. + for { + select { + case <-resp.Context().Done(): + return nil + case rtx, ok := <-ch: + if !ok { + return nil + } + if err := resp.Send(rtx); err != nil { + return err + } + } + } +} + func (s *SqlStreamer) GetBlock(ctx context.Context, id *walletrpc.BlockID) (*walletrpc.CompactBlock, error) { if id.Height == 0 && id.Hash == nil { return nil, ErrUnspecified diff --git a/walletrpc/service.pb.go b/walletrpc/service.pb.go index f73ee41..e8c4b25 100644 --- a/walletrpc/service.pb.go +++ b/walletrpc/service.pb.go @@ -694,6 +694,8 @@ type CompactTxStreamerClient interface { SendTransaction(ctx context.Context, in *RawTransaction, opts ...grpc.CallOption) (*SendResponse, error) // t-Address support GetAddressTxids(ctx context.Context, in *TransparentAddressBlockFilter, opts ...grpc.CallOption) (CompactTxStreamer_GetAddressTxidsClient, error) + // Mempool + GetMempoolStream(ctx context.Context, in *Empty, opts ...grpc.CallOption) (CompactTxStreamer_GetMempoolStreamClient, error) // Misc GetLightdInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*LightdInfo, error) GetCoinsupply(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Coinsupply, error) @@ -807,6 +809,38 @@ func (x *compactTxStreamerGetAddressTxidsClient) Recv() (*RawTransaction, error) return m, nil } +func (c *compactTxStreamerClient) GetMempoolStream(ctx context.Context, in *Empty, opts ...grpc.CallOption) (CompactTxStreamer_GetMempoolStreamClient, error) { + stream, err := c.cc.NewStream(ctx, &_CompactTxStreamer_serviceDesc.Streams[2], "/cash.z.wallet.sdk.rpc.CompactTxStreamer/GetMempoolStream", opts...) + if err != nil { + return nil, err + } + x := &compactTxStreamerGetMempoolStreamClient{stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +type CompactTxStreamer_GetMempoolStreamClient interface { + Recv() (*RawTransaction, error) + grpc.ClientStream +} + +type compactTxStreamerGetMempoolStreamClient struct { + grpc.ClientStream +} + +func (x *compactTxStreamerGetMempoolStreamClient) Recv() (*RawTransaction, error) { + m := new(RawTransaction) + if err := x.ClientStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + func (c *compactTxStreamerClient) GetLightdInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*LightdInfo, error) { out := new(LightdInfo) err := c.cc.Invoke(ctx, "/cash.z.wallet.sdk.rpc.CompactTxStreamer/GetLightdInfo", in, out, opts...) @@ -835,6 +869,8 @@ type CompactTxStreamerServer interface { SendTransaction(context.Context, *RawTransaction) (*SendResponse, error) // t-Address support GetAddressTxids(*TransparentAddressBlockFilter, CompactTxStreamer_GetAddressTxidsServer) error + // Mempool + GetMempoolStream(*Empty, CompactTxStreamer_GetMempoolStreamServer) error // Misc GetLightdInfo(context.Context, *Empty) (*LightdInfo, error) GetCoinsupply(context.Context, *Empty) (*Coinsupply, error) @@ -862,6 +898,9 @@ func (*UnimplementedCompactTxStreamerServer) SendTransaction(ctx context.Context func (*UnimplementedCompactTxStreamerServer) GetAddressTxids(req *TransparentAddressBlockFilter, srv CompactTxStreamer_GetAddressTxidsServer) error { return status.Errorf(codes.Unimplemented, "method GetAddressTxids not implemented") } +func (*UnimplementedCompactTxStreamerServer) GetMempoolStream(req *Empty, srv CompactTxStreamer_GetMempoolStreamServer) error { + return status.Errorf(codes.Unimplemented, "method GetMempoolStream not implemented") +} func (*UnimplementedCompactTxStreamerServer) GetLightdInfo(ctx context.Context, req *Empty) (*LightdInfo, error) { return nil, status.Errorf(codes.Unimplemented, "method GetLightdInfo not implemented") } @@ -987,6 +1026,27 @@ func (x *compactTxStreamerGetAddressTxidsServer) Send(m *RawTransaction) error { return x.ServerStream.SendMsg(m) } +func _CompactTxStreamer_GetMempoolStream_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(Empty) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(CompactTxStreamerServer).GetMempoolStream(m, &compactTxStreamerGetMempoolStreamServer{stream}) +} + +type CompactTxStreamer_GetMempoolStreamServer interface { + Send(*RawTransaction) error + grpc.ServerStream +} + +type compactTxStreamerGetMempoolStreamServer struct { + grpc.ServerStream +} + +func (x *compactTxStreamerGetMempoolStreamServer) Send(m *RawTransaction) error { + return x.ServerStream.SendMsg(m) +} + func _CompactTxStreamer_GetLightdInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(Empty) if err := dec(in); err != nil { @@ -1063,6 +1123,11 @@ var _CompactTxStreamer_serviceDesc = grpc.ServiceDesc{ Handler: _CompactTxStreamer_GetAddressTxids_Handler, ServerStreams: true, }, + { + StreamName: "GetMempoolStream", + Handler: _CompactTxStreamer_GetMempoolStream_Handler, + ServerStreams: true, + }, }, Metadata: "service.proto", } diff --git a/walletrpc/service.proto b/walletrpc/service.proto index b5e0086..eae3f36 100644 --- a/walletrpc/service.proto +++ b/walletrpc/service.proto @@ -87,6 +87,13 @@ service CompactTxStreamer { // t-Address support rpc GetAddressTxids(TransparentAddressBlockFilter) returns (stream RawTransaction) {} + // Mempool + // Return a stream of current mempool transactions as full RawTransactions. The stream stays + // open while there are mempool transactions and is closed when a new block is mined, at which + // point the client re-syncs the block and reconnects. Full RawTransactions (not CompactTx) are + // required so wallets can read shielded memos of 0-confirmation transactions (e.g. chat). + rpc GetMempoolStream(Empty) returns (stream RawTransaction) {} + // Misc rpc GetLightdInfo(Empty) returns (LightdInfo) {} rpc GetCoinsupply(Empty) returns (Coinsupply) {}