This repository has been archived by the owner on Jun 25, 2024. It is now read-only.
forked from lightninglabs/pool
-
Notifications
You must be signed in to change notification settings - Fork 0
/
manager.go
2039 lines (1768 loc) · 64.1 KB
/
manager.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package account
import (
"context"
"errors"
"fmt"
"math"
"strings"
"sync"
"time"
"github.com/btcsuite/btcd/blockchain"
"github.com/btcsuite/btcd/btcec"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
"github.com/btcsuite/btcutil/txsort"
"github.com/btcsuite/btcwallet/wallet"
"github.com/btcsuite/btcwallet/wallet/txrules"
"github.com/btcsuite/btcwallet/wtxmgr"
"github.com/getvoltage/pool/account/watcher"
"github.com/getvoltage/pool/poolscript"
"github.com/lightninglabs/lndclient"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/input"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
"github.com/lightningnetwork/lnd/lnwallet/chanfunding"
)
const (
// minConfs and maxConfs represent the thresholds at both extremes for
// valid number of confirmations on an account before it is considered
// open.
minConfs = 3
maxConfs = 6
// MinAccountValue is the minimum value for an account output in
// satoshis.
MinAccountValue btcutil.Amount = 100000
// minAccountExpiry and maxAccountExpiry represent the thresholds at
// both extremes for valid account expirations.
minAccountExpiry = 144 // One day worth of blocks.
maxAccountExpiry = 144 * 365 // A year worth of blocks.
)
var (
// errTxNotFound is an error returned when we attempt to locate a
// transaction but we are unable to find it.
errTxNotFound = errors.New("transaction not found")
)
// witnessType denotes the possible witness types of an account.
type witnessType uint8
const (
// expiryWitness is the type used for a witness taking the expiration
// path of an account.
expiryWitness witnessType = iota
// multiSigWitness is the type used for a witness taking the multi-sig
// path of an account.
multiSigWitness
)
// spendPackage tracks useful information regarding an account spend.
type spendPackage struct {
// tx is the spending transaction of the account.
tx *wire.MsgTx
// accountInputIdx is the index of the account input in the spending
// transaction.
accountInputIdx int
// witnessScript is the witness script of the account input being spent.
witnessScript []byte
// ourSig is our signature of the spending transaction above. If the
// spend is taking the multi-sig path, then the auctioneer's signature
// will be required as well for a valid spend.
ourSig []byte
}
// ManagerConfig contains all of the required dependencies for the Manager to
// carry out its duties.
type ManagerConfig struct {
// Store is responsible for storing and retrieving account information
// reliably.
Store Store
// Auctioneer provides us with the different ways we are able to
// communicate with our auctioneer during the process of
// opening/closing/modifying accounts.
Auctioneer Auctioneer
// Wallet handles all of our on-chain transaction interaction, whether
// that is deriving keys, creating transactions, etc.
Wallet lndclient.WalletKitClient
// Signer is responsible for deriving shared secrets for accounts
// between the trader and auctioneer and signing account-related
// transactions.
Signer lndclient.SignerClient
// ChainNotifier is responsible for requesting confirmation and spend
// notifications for accounts.
ChainNotifier lndclient.ChainNotifierClient
// TxSource is a source that provides us with transactions previously
// broadcast by us.
TxSource TxSource
// TxFeeEstimator is an estimator that can calculate the total on-chain
// fees to send to an account output.
TxFeeEstimator TxFeeEstimator
// TxLabelPrefix is set, then all transactions the account manager
// makes will use this string as a prefix for added transaction labels.
TxLabelPrefix string
}
// Manager is responsible for the management of accounts on-chain.
type Manager struct {
started sync.Once
stopped sync.Once
cfg ManagerConfig
watcher *watcher.Watcher
// pendingBatchMtx guards access to any database calls involving pending
// batches. This is mostly used to prevent race conditions when handling
// multiple accounts spends as part of a batch that we didn't receive a
// Finalize message for.
pendingBatchMtx sync.Mutex
// reservationMtx prevents a trader from attempting to have more than
// once active reservation at a time when creating new accounts. This is
// done to ensure an account picks up the correct reservation once its
// time to fund it.
reservationMtx sync.Mutex
wg sync.WaitGroup
quit chan struct{}
}
// NewManager instantiates a new Manager backed by the given config.
func NewManager(cfg *ManagerConfig) *Manager {
m := &Manager{
cfg: *cfg,
quit: make(chan struct{}),
}
m.watcher = watcher.New(&watcher.Config{
ChainNotifier: cfg.ChainNotifier,
HandleAccountConf: m.handleAccountConf,
HandleAccountSpend: m.handleAccountSpend,
HandleAccountExpiry: m.handleAccountExpiry,
})
return m
}
// Start resumes all account on-chain operation after a restart.
func (m *Manager) Start() error {
var err error
m.started.Do(func() {
err = m.start()
})
return err
}
// start resumes all account on-chain operation after a restart.
func (m *Manager) start() error {
ctx := context.Background()
// We'll start by resuming all of our accounts. This requires the
// watcher to be started first.
if err := m.watcher.Start(); err != nil {
return err
}
// Then, we'll resume all complete accounts, followed by partial
// accounts. If we were to do it the other way around, we'd resume
// partial accounts twice.
accounts, err := m.cfg.Store.Accounts()
if err != nil {
return fmt.Errorf("unable to retrieve accounts: %v", err)
}
for _, account := range accounts {
// Try to resume the account now.
//
// TODO(guggero): Refactor this to extract the init/funding
// part so we properly abandon the account if it fails before
// publishing the TX instead of trying to re-fund on startup.
err := m.resumeAccount(
ctx, account, true, false, DefaultFundingConfTarget,
)
if err != nil {
return fmt.Errorf("unable to resume account %x: %v",
account.TraderKey.PubKey.SerializeCompressed(),
err)
}
}
return nil
}
// Stop safely stops any ongoing operations within the Manager.
func (m *Manager) Stop() {
m.stopped.Do(func() {
m.watcher.Stop()
close(m.quit)
m.wg.Wait()
})
}
// QuoteAccount returns the expected fee rate and total miner fee to send to an
// account funding output with the given confTarget.
func (m *Manager) QuoteAccount(ctx context.Context, value btcutil.Amount,
confTarget uint32) (chainfee.SatPerKWeight, btcutil.Amount, error) {
// First, make sure we have a valid amount to create the account. We
// need to ask the auctioneer for the maximum as it dynamically defines
// that value.
terms, err := m.cfg.Auctioneer.Terms(ctx)
if err != nil {
return 0, 0, fmt.Errorf("could not query auctioneer terms: %v",
err)
}
err = validateAccountValue(value, terms.MaxAccountValue)
if err != nil {
return 0, 0, err
}
// Now calculate the estimated fee rate from the confTarget.
feeRate, err := m.cfg.Wallet.EstimateFee(ctx, int32(confTarget))
if err != nil {
return 0, 0, fmt.Errorf("error estimating fee rate: %v", err)
}
// Then calculate the total fee to pay. This asks lnd to create a full
// transaction to spend to a P2WSH output. If not enough confirmed funds
// are available in the wallet, this will return an error.
totalMinerFee, err := m.cfg.TxFeeEstimator.EstimateFeeToP2WSH(
ctx, value, int32(confTarget),
)
if err != nil {
return 0, 0, fmt.Errorf("error estimating total on-chain fee: "+
"%v", err)
}
return feeRate, totalMinerFee, nil
}
// InitAccount handles a request to create a new account with the provided
// parameters.
func (m *Manager) InitAccount(ctx context.Context, value btcutil.Amount,
expiry, bestHeight, confTarget uint32) (*Account, error) {
// We'll make sure to acquire the reservation lock throughout the
// account funding process to ensure we use the same reservation, as
// only one can be active per trader LSAT.
m.reservationMtx.Lock()
defer m.reservationMtx.Unlock()
// First, make sure we have a valid amount to create the account. We
// need to ask the auctioneer for the maximum as it dynamically defines
// that value.
terms, err := m.cfg.Auctioneer.Terms(ctx)
if err != nil {
return nil, fmt.Errorf("could not query auctioneer terms: %v",
err)
}
err = validateAccountParams(
value, terms.MaxAccountValue, expiry, bestHeight,
)
if err != nil {
return nil, err
}
// Let's make sure our wallet contains enough coins to fund the account
// before we reserve any resources. We ask lnd to create a transaction
// to send the given account value in dry-run mode. This makes sure we
// actually have some UTXOs to fund the account as this method returns
// an error on insufficient balance. Unfortunately it currently only
// supports estimating the total fee for a transaction using a
// confirmation target. We'll want to add sats/vByte as well as soon as
// the API allows it.
totalMinerFee, err := m.cfg.TxFeeEstimator.EstimateFeeToP2WSH(
ctx, value, int32(confTarget),
)
if err != nil {
return nil, fmt.Errorf("error estimating on-chain fee: %v", err)
}
// A by-product of the balance check is the total fee we'd need to pay
// so we might as well log it here.
log.Infof("Estimated total chain fee of %v for new account with "+
"value=%v, conf_target=%v", totalMinerFee, value, confTarget)
// We'll start by deriving a key for ourselves that we'll use in our
// 2-of-2 multi-sig construction.
keyDesc, err := m.cfg.Wallet.DeriveNextKey(
ctx, int32(poolscript.AccountKeyFamily),
)
if err != nil {
return nil, err
}
// With our key obtained, we'll reserve an account with our auctioneer,
// who will provide us with their base key and our initial per-batch
// key.
reservation, err := m.cfg.Auctioneer.ReserveAccount(
ctx, value, expiry, keyDesc.PubKey,
)
if err != nil {
return nil, err
}
// We'll also need to compute a shared secret based on both base keys
// (the trader and auctioneer's) to ensure only they are able to
// successfully identify every past/future output of the account.
secret, err := m.cfg.Signer.DeriveSharedKey(
ctx, reservation.AuctioneerKey, &keyDesc.KeyLocator,
)
if err != nil {
return nil, err
}
// With all of the details gathered, we'll persist our intent to create
// an account to disk and proceed to fund it and wait for its
// confirmation.
account := &Account{
Value: value,
Expiry: expiry,
TraderKey: keyDesc,
AuctioneerKey: reservation.AuctioneerKey,
BatchKey: reservation.InitialBatchKey,
Secret: secret,
State: StateInitiated,
HeightHint: bestHeight,
}
if err := m.cfg.Store.AddAccount(account); err != nil {
return nil, err
}
log.Infof("Creating new account %x of %v that expires at height %v",
keyDesc.PubKey.SerializeCompressed(), value, expiry)
err = m.resumeAccount(ctx, account, false, false, confTarget)
if err != nil {
return nil, err
}
return account, nil
}
// WatchMatchedAccounts resumes accounts that were just matched in a batch and
// are expecting the batch transaction to confirm as their next account output.
// This will cancel all previous spend and conf watchers of all accounts
// involved in the batch.
func (m *Manager) WatchMatchedAccounts(ctx context.Context,
matchedAccounts []*btcec.PublicKey) error {
for _, matchedAccount := range matchedAccounts {
acct, err := m.cfg.Store.Account(matchedAccount)
if err != nil {
return fmt.Errorf("error reading account %x: %v",
matchedAccount.SerializeCompressed(), err)
}
// The account was just involved in a batch. That means our
// account output was spent by a batch transaction. Since we
// know that a batch transaction cannot simply be rolled back or
// replaced without us being involved, we know that the batch TX
// will eventually confirm. To handle the case where an account
// is involved in multiple consecutive batches that are all
// unconfirmed, we make sure we only track the latest state by
// canceling all previous spend and confirmation watchers. We
// then only watch the latest batch and once it confirms, create
// a new spend watcher on that.
m.watcher.CancelAccountSpend(matchedAccount)
m.watcher.CancelAccountConf(matchedAccount)
// After taking part in a batch, the account is either pending
// closed because it was used up or pending batch update because
// it was recreated. Either way, let's resume it now by creating
// the appropriate watchers again.
err = m.resumeAccount(ctx, acct, false, false, 0)
if err != nil {
return fmt.Errorf("error resuming account %x: %v",
matchedAccount.SerializeCompressed(), err)
}
}
return nil
}
// maybeBroadcastTx attempts to broadcast the transaction only if all of its
// inputs have been signed for.
func (m *Manager) maybeBroadcastTx(ctx context.Context, tx *wire.MsgTx,
label string) error {
// If any of the transaction inputs aren't signed, don't broadcast.
for _, txIn := range tx.TxIn {
if len(txIn.Witness) == 0 && len(txIn.SignatureScript) == 0 {
return nil
}
}
return m.cfg.Wallet.PublishTransaction(ctx, tx, label)
}
// resumeAccount performs different operations based on the account's state.
// This method serves as a way to consolidate the logic of resuming accounts on
// startup and during normal operation.
func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint
onRestart bool, onRecovery bool, fundingConfTarget uint32) error {
accountOutput, err := account.Output()
if err != nil {
return fmt.Errorf("unable to construct account output: %v", err)
}
switch account.State {
// In StateInitiated, we'll attempt to fund our account.
case StateInitiated:
// If we're resuming the account from a restart, we'll want to
// make sure we haven't created and broadcast a transaction for
// this account already, so we'll inspect our TxSource to do so.
var (
accountTx *wire.MsgTx
createTx = true
)
if onRestart || onRecovery {
tx, err := m.locateTxByOutput(
ctx, accountOutput, account.LatestTx,
)
switch err {
// If we do find one, we can rebroadcast it.
case nil:
accountTx = tx
createTx = false
// If we don't, we'll need to create one.
case errTxNotFound:
// If lnd doesn't know a transaction that sends
// to the account output, it could be that it
// was never published or it never confirmed.
// In that case the funds should be SAFU and can
// be double spent. We don't need to try a
// recovery in that case. And we certainly don't
// want to send funds again, so we exit here.
if onRecovery {
state := StateCanceledAfterRecovery
err := m.cfg.Store.UpdateAccount(
account, StateModifier(state),
)
if err != nil {
return fmt.Errorf("account "+
"funding TX not found "+
"but was unable to "+
"update account to "+
"state recovery "+
"failed: %v", err)
}
return fmt.Errorf("account funding "+
"TX with output %x not found",
accountOutput.PkScript)
}
default:
return fmt.Errorf("unable to locate output "+
"%x: %v", accountOutput.PkScript, err)
}
}
if createTx {
// We need a static sat/vByte value for the fee in
// SendOutputs, so we use the stored targetConf of the
// account to estimate it.
feeSatPerKw, err := m.cfg.Wallet.EstimateFee(
ctx, int32(fundingConfTarget),
)
if err != nil {
return err
}
// If we have a label prefix, then we'll apply that now
// and also attach some additional meta data.
acctKey := account.TraderKey.PubKey.SerializeCompressed()
contextLabel := fmt.Sprintf(" poold -- "+
"AccountCreation(acct_key=%x)", acctKey)
label := makeTxnLabel(m.cfg.TxLabelPrefix, contextLabel)
// TODO(wilmer): Expose manual controls to bump fees.
tx, err := m.cfg.Wallet.SendOutputs(
ctx, []*wire.TxOut{accountOutput}, feeSatPerKw,
label,
)
if err != nil {
return err
}
accountTx = tx
log.Infof("Funded new account %x with transaction %v",
account.TraderKey.PubKey.SerializeCompressed(),
tx.TxHash())
}
// With the transaction obtained, we'll locate the index of our
// account output in the transaction to obtain our account
// outpoint and store it to disk. This will be the main way we
// identify our accounts, and is also required to watch for its
// spend.
outputIndex, ok := poolscript.LocateOutputScript(
accountTx, accountOutput.PkScript,
)
if !ok {
return fmt.Errorf("transaction %v does not include "+
"expected script %x", accountTx.TxHash(),
accountOutput.PkScript)
}
op := wire.OutPoint{Hash: accountTx.TxHash(), Index: outputIndex}
err := m.cfg.Store.UpdateAccount(
account, StateModifier(StatePendingOpen),
OutPointModifier(op), LatestTxModifier(accountTx),
)
if err != nil {
return err
}
fallthrough
// In StatePendingOpen, we should already have broadcast a funding
// transaction for the account, so the most we can do is attempt to
// rebroadcast it and wait for its confirmation.
case StatePendingOpen:
// If we're resuming from a restart, we'll have to locate the
// transaction in our TxSource by its hash. We should definitely
// find one in this state, so if we don't, that would indicate
// something has gone wrong.
if onRestart {
accountTx := account.LatestTx
// Since we store the latest account modification TX in
// the account itself, we don't need to rely on lnd
// keeping track of all our TXns anymore. If what we
// have in the DB is correct, we can just re-broadcast
// that TX.
if accountTx == nil ||
accountTx.TxHash() != account.OutPoint.Hash {
var err error
accountTx, err = m.locateTxByHash(
ctx, account.OutPoint.Hash,
)
if err != nil {
return fmt.Errorf("unable to locate "+
"transaction %v: %v",
account.OutPoint.Hash, err)
}
}
acctKey := account.TraderKey.PubKey.SerializeCompressed()
contextLabel := fmt.Sprintf(" poold -- "+
"AccountCreation(acct_key=%x)", acctKey)
label := makeTxnLabel(m.cfg.TxLabelPrefix, contextLabel)
err = m.maybeBroadcastTx(ctx, accountTx, label)
if err != nil {
return err
}
}
// Send the account parameters over to the auctioneer so that
// they're also aware of the account.
err := m.cfg.Auctioneer.InitAccount(ctx, account)
if err != nil {
return err
}
terms, err := m.cfg.Auctioneer.Terms(ctx)
if err != nil {
return fmt.Errorf("could not query auctioneer terms: "+
"%v", err)
}
// Proceed to watch for the account on-chain.
numConfs := NumConfsForValue(
account.Value, terms.MaxAccountValue,
)
log.Infof("Waiting for %v confirmation(s) of account %x",
numConfs, account.TraderKey.PubKey.SerializeCompressed())
err = m.watcher.WatchAccountConf(
account.TraderKey.PubKey, account.OutPoint.Hash,
accountOutput.PkScript, numConfs, account.HeightHint,
)
if err != nil {
return fmt.Errorf("unable to watch for confirmation: "+
"%v", err)
}
// In StatePendingUpdate or StatePendingBatch, we've processed an
// account update due to either a matched order or trader modification,
// so we'll need to wait for its confirmation. Once it confirms,
// handleAccountConf will take care of the rest of the flow.
//
// TODO(wilmer): Handle restart case where the client shuts down after
// the modification has been reflected on-disk, but the auctioneer's
// signature hasn't been received.
//
// TODO(guggero): Handle the case of a malicious auctioneer that
// replaces batch A with a batch A' that contains none of our accounts
// and would therefore not be noticed by us. The account would stay
// pending forever in that case.
case StatePendingUpdate, StatePendingBatch:
// We need to know the maximum account value to scale the number
// of confirmations the same way the auctioneer does to avoid
// getting the state out of sync.
terms, err := m.cfg.Auctioneer.Terms(ctx)
if err != nil {
return fmt.Errorf("could not query auctioneer terms: "+
"%v", err)
}
numConfs := NumConfsForValue(
account.Value, terms.MaxAccountValue,
)
log.Infof("Waiting for %v confirmation(s) of account %x",
numConfs, account.TraderKey.PubKey.SerializeCompressed())
err = m.watcher.WatchAccountConf(
account.TraderKey.PubKey, account.OutPoint.Hash,
accountOutput.PkScript, numConfs, account.HeightHint,
)
if err != nil {
return fmt.Errorf("unable to watch for confirmation: "+
"%v", err)
}
// Only subscribe to auction updates for this account if it's in
// the pending batch state, to allow traders to participate in
// consecutive batches. This isn't necessary for the pending
// update state, as that state is ineligible for batch
// execution.
if account.State == StatePendingBatch {
err = m.cfg.Auctioneer.StartAccountSubscription(
ctx, account.TraderKey,
)
if err != nil {
return fmt.Errorf("unable to subscribe for "+
"account updates: %v", err)
}
}
// In StateOpen, the funding transaction for the account has already
// confirmed, so we only need to watch for its spend and expiration and
// register for account updates.
case StateOpen:
if err := m.handleStateOpen(ctx, account); err != nil {
return err
}
// In StateExpiredPendingUpdate, the account expired while having a
// pending update. To make sure the account can be renewed, we'll wait
// for the pending update to confirm and transition the account to
// StateExpired then.
case StateExpiredPendingUpdate:
terms, err := m.cfg.Auctioneer.Terms(ctx)
if err != nil {
return fmt.Errorf("could not query auctioneer terms: "+
"%v", err)
}
numConfs := NumConfsForValue(
account.Value, terms.MaxAccountValue,
)
log.Infof("Waiting for %v confirmation(s) of expired account %x",
numConfs, account.TraderKey.PubKey.SerializeCompressed())
err = m.watcher.WatchAccountConf(
account.TraderKey.PubKey, account.OutPoint.Hash,
accountOutput.PkScript, numConfs, account.HeightHint,
)
if err != nil {
return fmt.Errorf("unable to watch for confirmation: "+
"%v", err)
}
// In StateExpired, we'll wait for the account to be spent so that we
// can detect whether its been closed or renewed.
case StateExpired:
log.Infof("Watching expired account %x for spend",
account.TraderKey.PubKey.SerializeCompressed())
err = m.watcher.WatchAccountSpend(
account.TraderKey.PubKey, account.OutPoint,
accountOutput.PkScript, account.HeightHint,
)
if err != nil {
return fmt.Errorf("unable to watch for spend: %v", err)
}
// In StatePendingClosed, we'll wait for the account's closing
// transaction to confirm so that we can transition the account to its
// final state.
case StatePendingClosed:
acctKey := account.TraderKey.PubKey.SerializeCompressed()
contextLabel := fmt.Sprintf(" poold -- "+
"AccountClosure(acct_key=%x)", acctKey)
label := makeTxnLabel(m.cfg.TxLabelPrefix, contextLabel)
err := m.maybeBroadcastTx(ctx, account.LatestTx, label)
if err != nil {
return err
}
log.Infof("Watching account %x for spend",
account.TraderKey.PubKey.SerializeCompressed())
err = m.watcher.WatchAccountSpend(
account.TraderKey.PubKey, account.OutPoint,
accountOutput.PkScript, account.HeightHint,
)
if err != nil {
return fmt.Errorf("unable to watch for spend: %v", err)
}
// If the account has already been closed or canceled, there's nothing
// to be done.
case StateClosed, StateCanceledAfterRecovery:
break
default:
return fmt.Errorf("unhandled account state %v", account.State)
}
return nil
}
// locateTxByOutput locates a transaction from the Manager's TxSource by one of
// its outputs. If a transaction is not found containing the output, then
// errTxNotFound is returned.
func (m *Manager) locateTxByOutput(ctx context.Context,
output *wire.TxOut, fullTx *wire.MsgTx) (*wire.MsgTx, error) {
// We now store the full raw transaction of the last modification. We
// can just use that if available. If for some reason that TX doesn't
// contain our current outpoint, we fall back to the previous behavior.
if fullTx != nil {
idx, ok := poolscript.LocateOutputScript(fullTx, output.PkScript)
if ok && fullTx.TxOut[idx].Value == output.Value {
return fullTx, nil
}
}
// Get all transactions, starting from block 0 and including unconfirmed
// TXes (end block = -1).
txs, err := m.cfg.TxSource.ListTransactions(ctx, 0, -1)
if err != nil {
return nil, err
}
for _, tx := range txs {
idx, ok := poolscript.LocateOutputScript(tx.Tx, output.PkScript)
if !ok {
continue
}
if tx.Tx.TxOut[idx].Value == output.Value {
return tx.Tx, nil
}
}
return nil, errTxNotFound
}
// locateTxByHash locates a transaction from the Manager's TxSource by its hash.
// If the transaction is not found, then errTxNotFound is returned.
func (m *Manager) locateTxByHash(ctx context.Context,
hash chainhash.Hash) (*wire.MsgTx, error) {
// Get all transactions, starting from block 0 and including unconfirmed
// TXes (end block = -1).
txs, err := m.cfg.TxSource.ListTransactions(ctx, 0, -1)
if err != nil {
return nil, err
}
for _, tx := range txs {
if tx.Tx.TxHash() == hash {
return tx.Tx, nil
}
}
return nil, errTxNotFound
}
// handleStateOpen performs the necessary operations for accounts found in
// StateOpen.
func (m *Manager) handleStateOpen(ctx context.Context, account *Account) error {
var traderKey [33]byte
copy(traderKey[:], account.TraderKey.PubKey.SerializeCompressed())
log.Infof("Watching spend of %v for account %x", account.OutPoint,
traderKey)
accountOutput, err := account.Output()
if err != nil {
return err
}
err = m.watcher.WatchAccountSpend(
account.TraderKey.PubKey, account.OutPoint,
accountOutput.PkScript, account.HeightHint,
)
if err != nil {
return fmt.Errorf("unable to watch for spend: %v", err)
}
err = m.watcher.WatchAccountExpiration(
account.TraderKey.PubKey, account.Expiry,
)
if err != nil {
return fmt.Errorf("unable to watch for expiration: %v", err)
}
// Now that we have an open account, subscribe for updates to it to the
// server. We subscribe for the account instead of the individual orders
// because all signing operations will need to be executed on an account
// level anyway. And we might end up executing multiple orders for the
// same account in one batch. The messages from the server are received
// and dispatched to the correct manager by the rpcServer.
err = m.cfg.Auctioneer.StartAccountSubscription(ctx, account.TraderKey)
if err != nil {
return fmt.Errorf("unable to subscribe for account updates: %v",
err)
}
return nil
}
// handleAccountConf takes the necessary steps after detecting the confirmation
// of an account on-chain.
func (m *Manager) handleAccountConf(traderKey *btcec.PublicKey,
confDetails *chainntnfs.TxConfirmation) error {
account, err := m.cfg.Store.Account(traderKey)
if err != nil {
return err
}
log.Infof("Account %x is now confirmed at height %v!",
traderKey.SerializeCompressed(), confDetails.BlockHeight)
// The new state we'll transition to depends on the account's current
// state.
var newState State
switch account.State {
// Any pending states will transition to their confirmed state.
case StatePendingOpen, StatePendingUpdate, StatePendingBatch:
newState = StateOpen
// An expired account with a pending update that has now confirmed will
// transition to the confirmed expired case, allowing a trader to renew
// their account.
case StateExpiredPendingUpdate:
newState = StateExpired
default:
return fmt.Errorf("unhandled state %v after confirmation",
account.State)
}
// Update the account's state and proceed with the rest of the flow.
mods := []Modifier{
StateModifier(newState),
HeightHintModifier(confDetails.BlockHeight),
}
if err := m.cfg.Store.UpdateAccount(account, mods...); err != nil {
return err
}
return m.handleStateOpen(context.Background(), account)
}
// handleAccountSpend handles the different spend paths of an account. If an
// account is spent by the expiration path, it'll always be marked as closed
// thereafter. If it is spent by the cooperative path with the auctioneer, then
// the account will only remain open if the spending transaction recreates the
// account with the expected next account script. Otherwise, it is also marked
// as closed. In case of multiple consecutive batches with the same account, we
// only track the spend of the latest batch, after it confirmed. So the account
// output in the spend transaction should always match our database state if
// it was a cooperative spend.
func (m *Manager) handleAccountSpend(traderKey *btcec.PublicKey,
spendDetails *chainntnfs.SpendDetail) error {
account, err := m.cfg.Store.Account(traderKey)
if err != nil {
return err
}
// We'll need to perform different operations based on the witness of
// the spending input of the account.
spendTx := spendDetails.SpendingTx
spendWitness := spendTx.TxIn[spendDetails.SpenderInputIndex].Witness
switch {
// If the witness is for a spend of the account expiration path, then
// we'll mark the account as closed as the account has expired and all
// the funds have been withdrawn.
case poolscript.IsExpirySpend(spendWitness):
break
// If the witness is for a multi-sig spend, then either an order by the
// trader was matched, or the account was closed. If it was closed, then
// the account output shouldn't have been recreated.
case poolscript.IsMultiSigSpend(spendWitness):
// If there's a pending batch which has yet to be completed,
// we'll mark it as so now. This can happen if the trader is not
// connected to the auctioneer when the auctioneer sends them
// the finalize message.
//
// We'll acquire the pending batch lock to ensure that there
// aren't multiple handleAccountSpend threads (in the case of
// multiple accounts participating in a batch) attempting to
// mark the same batch as complete and prevent entering into an
// erroneous state.
m.pendingBatchMtx.Lock()
err := m.cfg.Store.PendingBatch()
switch err {
// If there's no pending batch, we can proceed as normal.
case ErrNoPendingBatch:
break
// If there is, we'll commit it and refresh the account state.
case nil:
if err := m.cfg.Store.MarkBatchComplete(); err != nil {
m.pendingBatchMtx.Unlock()
return err
}
account, err = m.cfg.Store.Account(traderKey)
if err != nil {
m.pendingBatchMtx.Unlock()
return err
}
default:
m.pendingBatchMtx.Unlock()
return err
}
m.pendingBatchMtx.Unlock()
// An account cannot be spent without our knowledge, so we'll
// assume we always persist account updates before a broadcast
// of the spending transaction. Therefore, since we should
// already have the updates applied, we can just look for our
// current output in the transaction.
accountOutput, err := account.Output()
if err != nil {
return err
}
_, ok := poolscript.LocateOutputScript(
spendTx, accountOutput.PkScript,
)
if ok {
// Proceed with the rest of the flow. We won't send to
// the account output again, so we don't need to set
// a valid conf target.
return m.resumeAccount(
context.Background(), account, false, false, 0,
)
}
default:
return fmt.Errorf("unknown spend witness %x", spendWitness)
}
log.Infof("Account %x has been closed on-chain with transaction %v",
account.TraderKey.PubKey.SerializeCompressed(), spendTx.TxHash())
// Write the spending transaction once again in case the one we
// previously broadcast was replaced with a higher fee one.
return m.cfg.Store.UpdateAccount(
account, StateModifier(StateClosed),
HeightHintModifier(uint32(spendDetails.SpendingHeight)),
LatestTxModifier(spendTx),
)
}
// handleAccountExpiry marks an account as expired within the database.
func (m *Manager) handleAccountExpiry(traderKey *btcec.PublicKey,
height uint32) error {
account, err := m.cfg.Store.Account(traderKey)
if err != nil {
return err