mirror of
https://github.com/lightninglabs/loop.git
synced 2026-08-13 12:33:03 +02:00
Merge pull request #890 from starius/unit-race
multi: fix race conditions and run unit tests with -race
This commit is contained in:
Boris Nagaev
GitHub
commit
75ad9b4649
31 changed files with 202 additions and 197 deletions
4
.github/workflows/main.yml
vendored
4
.github/workflows/main.yml
vendored
|
|
@ -101,7 +101,7 @@ jobs:
|
|||
go-version: '~${{ env.GO_VERSION }}'
|
||||
|
||||
- name: run unit tests
|
||||
run: make unit
|
||||
run: make unit-race
|
||||
|
||||
- name: run unit test with postgres
|
||||
run: make unit-postgres
|
||||
run: make unit-postgres-race
|
||||
|
|
|
|||
10
Makefile
10
Makefile
|
|
@ -95,14 +95,22 @@ clean:
|
|||
# TESTING
|
||||
# =======
|
||||
|
||||
unit:
|
||||
unit:
|
||||
@$(call print, "Running unit tests.")
|
||||
$(UNIT)
|
||||
|
||||
unit-race:
|
||||
@$(call print, "Running unit race tests.")
|
||||
$(UNIT) -race
|
||||
|
||||
unit-postgres:
|
||||
@$(call print, "Running unit tests with postgres.")
|
||||
$(UNIT) -tags=test_db_postgres
|
||||
|
||||
unit-postgres-race:
|
||||
@$(call print, "Running unit race tests with postgres.")
|
||||
$(UNIT) -race -tags=test_db_postgres
|
||||
|
||||
# =========
|
||||
# UTILITIES
|
||||
# =========
|
||||
|
|
|
|||
3
go.mod
3
go.mod
|
|
@ -9,7 +9,6 @@ require (
|
|||
github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c // indirect
|
||||
github.com/btcsuite/btcwallet v0.16.10-0.20241127094224-93c858b2ad63
|
||||
github.com/btcsuite/btcwallet/wtxmgr v1.5.4
|
||||
github.com/coreos/bbolt v1.3.3
|
||||
github.com/davecgh/go-spew v1.1.1
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0
|
||||
github.com/fortytw2/leaktest v1.3.0
|
||||
|
|
@ -35,6 +34,7 @@ require (
|
|||
github.com/ory/dockertest/v3 v3.10.0
|
||||
github.com/stretchr/testify v1.9.0
|
||||
github.com/urfave/cli v1.22.9
|
||||
go.etcd.io/bbolt v1.3.11
|
||||
golang.org/x/net v0.36.0
|
||||
google.golang.org/grpc v1.64.1
|
||||
google.golang.org/protobuf v1.34.2
|
||||
|
|
@ -164,7 +164,6 @@ require (
|
|||
github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
|
||||
github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 // indirect
|
||||
gitlab.com/yawning/bsaes.git v0.0.0-20190805113838-0a714cd429ec // indirect
|
||||
go.etcd.io/bbolt v1.3.11 // indirect
|
||||
go.etcd.io/etcd/api/v3 v3.5.12 // indirect
|
||||
go.etcd.io/etcd/client/pkg/v3 v3.5.12 // indirect
|
||||
go.etcd.io/etcd/client/v2 v2.305.12 // indirect
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -732,8 +732,6 @@ github.com/cockroachdb/datadriven v1.0.2 h1:H9MtNqVoVhvd9nCBwOyDjUEdZCREqbIdCJD9
|
|||
github.com/cockroachdb/datadriven v1.0.2/go.mod h1:a9RdTaap04u637JoCzcUoIcDmvwSUtcUFtT/C3kJlTU=
|
||||
github.com/containerd/continuity v0.3.0 h1:nisirsYROK15TAMVukJOUyGJjz4BNQJBVsNvAXZJ/eg=
|
||||
github.com/containerd/continuity v0.3.0/go.mod h1:wJEAIwKOm/pBZuBd0JmeTvnLquTB1Ag8espWhkykbPM=
|
||||
github.com/coreos/bbolt v1.3.3 h1:n6AiVyVRKQFNb6mJlwESEvvLoDyiTzXX7ORAUlkeBdY=
|
||||
github.com/coreos/bbolt v1.3.3/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
|
||||
github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
|
||||
github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
|
||||
github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
|
||||
|
|
|
|||
|
|
@ -41,18 +41,17 @@ type Manager struct {
|
|||
}
|
||||
|
||||
// NewInstantOutManager creates a new instantout manager.
|
||||
func NewInstantOutManager(cfg *Config) *Manager {
|
||||
func NewInstantOutManager(cfg *Config, height int32) *Manager {
|
||||
return &Manager{
|
||||
cfg: cfg,
|
||||
activeInstantOuts: make(map[lntypes.Hash]*FSM),
|
||||
blockEpochChan: make(chan int32),
|
||||
currentHeight: height,
|
||||
}
|
||||
}
|
||||
|
||||
// Run runs the instantout manager.
|
||||
func (m *Manager) Run(ctx context.Context, initChan chan struct{},
|
||||
height int32) error {
|
||||
|
||||
func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
|
||||
log.Debugf("Starting instantout manager")
|
||||
defer func() {
|
||||
log.Debugf("Stopping instantout manager")
|
||||
|
|
@ -62,7 +61,6 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{},
|
|||
defer cancel()
|
||||
|
||||
m.runCtx = runCtx
|
||||
m.currentHeight = height
|
||||
|
||||
err := m.recoverInstantOuts(runCtx)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package reservation
|
|||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
|
|
@ -72,13 +73,14 @@ func TestManager(t *testing.T) {
|
|||
require.NoError(t, err)
|
||||
|
||||
// We'll now expect a spend registration.
|
||||
spendReg := <-testContext.mockLnd.RegisterSpendChannel
|
||||
require.Equal(t, spendReg.PkScript, pkScript)
|
||||
var spendReg atomic.Pointer[test.SpendRegistration]
|
||||
spendReg.Store(<-testContext.mockLnd.RegisterSpendChannel)
|
||||
require.Equal(t, spendReg.Load().PkScript, pkScript)
|
||||
|
||||
go func() {
|
||||
// We'll expect a second spend registration.
|
||||
spendReg = <-testContext.mockLnd.RegisterSpendChannel
|
||||
require.Equal(t, spendReg.PkScript, pkScript)
|
||||
spendReg.Store(<-testContext.mockLnd.RegisterSpendChannel)
|
||||
require.Equal(t, spendReg.Load().PkScript, pkScript)
|
||||
}()
|
||||
|
||||
// We'll now try to lock the reservation.
|
||||
|
|
@ -90,7 +92,7 @@ func TestManager(t *testing.T) {
|
|||
require.Error(t, err)
|
||||
|
||||
testContext.mockLnd.SpendChannel <- &chainntnfs.SpendDetail{
|
||||
SpentOutPoint: spendReg.Outpoint,
|
||||
SpentOutPoint: spendReg.Load().Outpoint,
|
||||
}
|
||||
|
||||
// We'll now expect the reservation to be expired.
|
||||
|
|
|
|||
|
|
@ -423,7 +423,7 @@ func getTLSConfig(cfg *Config) (*tls.Config, *credentials.TransportCredentials,
|
|||
// If the certificate expired or it was outdated, delete it and the TLS
|
||||
// key and generate a new pair.
|
||||
if time.Now().After(parsedCert.NotAfter) {
|
||||
log.Info("TLS certificate is expired or outdated, " +
|
||||
infof("TLS certificate is expired or outdated, " +
|
||||
"removing old file then generating a new one")
|
||||
|
||||
err := os.Remove(cfg.TLSCertPath)
|
||||
|
|
@ -464,7 +464,7 @@ func loadCertWithCreate(cfg *Config) (tls.Certificate, *x509.Certificate,
|
|||
if !lnrpc.FileExists(cfg.TLSCertPath) &&
|
||||
!lnrpc.FileExists(cfg.TLSKeyPath) {
|
||||
|
||||
log.Infof("Generating TLS certificates...")
|
||||
infof("Generating TLS certificates...")
|
||||
certBytes, keyBytes, err := cert.GenCertPair(
|
||||
defaultSelfSignedOrganization, cfg.TLSExtraIPs,
|
||||
cfg.TLSExtraDomains, cfg.TLSDisableAutofill,
|
||||
|
|
@ -481,7 +481,7 @@ func loadCertWithCreate(cfg *Config) (tls.Certificate, *x509.Certificate,
|
|||
return tls.Certificate{}, nil, err
|
||||
}
|
||||
|
||||
log.Infof("Done generating TLS certificates")
|
||||
infof("Done generating TLS certificates")
|
||||
}
|
||||
|
||||
return cert.LoadCert(cfg.TLSCertPath, cfg.TLSKeyPath)
|
||||
|
|
|
|||
155
loopd/daemon.go
155
loopd/daemon.go
|
|
@ -12,7 +12,6 @@ import (
|
|||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/bbolt"
|
||||
proxy "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
|
||||
"github.com/lightninglabs/lndclient"
|
||||
"github.com/lightninglabs/loop"
|
||||
|
|
@ -33,6 +32,7 @@ import (
|
|||
"github.com/lightningnetwork/lnd/clock"
|
||||
"github.com/lightningnetwork/lnd/lntypes"
|
||||
"github.com/lightningnetwork/lnd/macaroons"
|
||||
"go.etcd.io/bbolt"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/protobuf/encoding/protojson"
|
||||
"gopkg.in/macaroon-bakery.v2/bakery"
|
||||
|
|
@ -169,11 +169,11 @@ func (d *Daemon) Start() error {
|
|||
// anything goes wrong now, we need to cleanly shut down again.
|
||||
startErr := d.startWebServers()
|
||||
if startErr != nil {
|
||||
log.Errorf("Error while starting daemon: %v", err)
|
||||
errorf("Error while starting daemon: %v", err)
|
||||
d.Stop()
|
||||
stopErr := <-d.ErrChan
|
||||
if stopErr != nil {
|
||||
log.Errorf("Error while stopping daemon: %v", stopErr)
|
||||
errorf("Error while stopping daemon: %v", stopErr)
|
||||
}
|
||||
return startErr
|
||||
}
|
||||
|
|
@ -253,7 +253,7 @@ func (d *Daemon) startWebServers() error {
|
|||
d.registerDebugServer()
|
||||
|
||||
// Next, start the gRPC server listening for HTTP/2 connections.
|
||||
log.Infof("Starting gRPC listener")
|
||||
infof("Starting gRPC listener")
|
||||
serverTLSCfg, restClientCreds, err := getTLSConfig(d.cfg)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not create gRPC server options: %v",
|
||||
|
|
@ -322,7 +322,7 @@ func (d *Daemon) startWebServers() error {
|
|||
|
||||
// A nil listener indicates REST is disabled.
|
||||
if d.restListener != nil {
|
||||
log.Infof("Starting REST proxy listener")
|
||||
infof("Starting REST proxy listener")
|
||||
|
||||
d.restServer = &http.Server{
|
||||
Handler: restHandler,
|
||||
|
|
@ -333,7 +333,7 @@ func (d *Daemon) startWebServers() error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Infof("REST proxy listening on %s",
|
||||
infof("REST proxy listening on %s",
|
||||
d.restListener.Addr())
|
||||
err := d.restServer.Serve(d.restListener)
|
||||
// ErrServerClosed is always returned when the proxy is
|
||||
|
|
@ -347,7 +347,7 @@ func (d *Daemon) startWebServers() error {
|
|||
}
|
||||
}()
|
||||
} else {
|
||||
log.Infof("REST proxy disabled")
|
||||
infof("REST proxy disabled")
|
||||
}
|
||||
|
||||
// Start the grpc server.
|
||||
|
|
@ -355,7 +355,7 @@ func (d *Daemon) startWebServers() error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Infof("RPC server listening on %s", d.grpcListener.Addr())
|
||||
infof("RPC server listening on %s", d.grpcListener.Addr())
|
||||
err = d.grpcServer.Serve(d.grpcListener)
|
||||
if err != nil && !errors.Is(err, grpc.ErrServerStopped) {
|
||||
// Notify the main error handler goroutine that
|
||||
|
|
@ -378,7 +378,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
loopdb.EnableExperimentalProtocol()
|
||||
}
|
||||
|
||||
log.Infof("Protocol version: %v", loopdb.CurrentProtocolVersion())
|
||||
infof("Protocol version: %v", loopdb.CurrentProtocolVersion())
|
||||
|
||||
// If no swap server is specified, use the default addresses for mainnet
|
||||
// and testnet.
|
||||
|
|
@ -404,18 +404,18 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
// on main context cancel. So we create it early and pass it down.
|
||||
d.mainCtx, d.mainCtxCancel = context.WithCancel(context.Background())
|
||||
|
||||
log.Infof("Swap server address: %v", d.cfg.Server.Host)
|
||||
infof("Swap server address: %v", d.cfg.Server.Host)
|
||||
|
||||
// Check if we need to migrate the database.
|
||||
if needSqlMigration(d.cfg) {
|
||||
log.Infof("Boltdb found, running migration")
|
||||
infof("Boltdb found, running migration")
|
||||
|
||||
err := migrateBoltdb(d.mainCtx, d.cfg)
|
||||
if err != nil {
|
||||
return fmt.Errorf("unable to migrate boltdb: %v", err)
|
||||
}
|
||||
|
||||
log.Infof("Successfully migrated boltdb")
|
||||
infof("Successfully migrated boltdb")
|
||||
}
|
||||
|
||||
// Now that we know where the database will live, we'll go ahead and
|
||||
|
|
@ -436,7 +436,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
swapDb,
|
||||
)
|
||||
if err != nil {
|
||||
log.Errorf("Cost migration failed: %v", err)
|
||||
errorf("Cost migration failed: %v", err)
|
||||
|
||||
return err
|
||||
}
|
||||
|
|
@ -446,6 +446,14 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
chainParams,
|
||||
)
|
||||
|
||||
// We need to know the current block height to properly initialize
|
||||
// managers.
|
||||
getInfo, err := d.lnd.Client.GetInfo(d.mainCtx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get current block height: %w", err)
|
||||
}
|
||||
blockHeight := getInfo.BlockHeight
|
||||
|
||||
// If we're running an asset client, we'll log something here.
|
||||
if d.assetClient != nil {
|
||||
getInfo, err := d.assetClient.GetInfo(
|
||||
|
|
@ -460,7 +468,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
d.lnd.NodePubkey)
|
||||
}
|
||||
|
||||
log.Infof("Using asset client with version %v", getInfo.Version)
|
||||
infof("Using asset client with version %v", getInfo.Version)
|
||||
}
|
||||
|
||||
// Create an instance of the loop client library.
|
||||
|
|
@ -507,7 +515,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
cleanupMacaroonStore := func() {
|
||||
err := db.Close()
|
||||
if err != nil {
|
||||
log.Errorf("Error closing macaroon store: %v", err)
|
||||
errorf("Error closing macaroon store: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -555,11 +563,11 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Info("Starting notification manager")
|
||||
infof("Starting notification manager")
|
||||
err := notificationManager.Run(d.mainCtx)
|
||||
if err != nil {
|
||||
d.internalErrChan <- err
|
||||
log.Errorf("Notification manager stopped: %v", err)
|
||||
errorf("Notification manager stopped: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
|
|
@ -580,7 +588,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
ChainParams: d.lnd.ChainParams,
|
||||
ChainNotifier: d.lnd.ChainNotifier,
|
||||
}
|
||||
staticAddressManager = address.NewManager(addrCfg)
|
||||
staticAddressManager = address.NewManager(addrCfg, int32(blockHeight))
|
||||
|
||||
// Static address deposit manager setup.
|
||||
depositStore := deposit.NewSqlStore(baseDb)
|
||||
|
|
@ -606,7 +614,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
ChainNotifier: d.lnd.ChainNotifier,
|
||||
Signer: d.lnd.Signer,
|
||||
}
|
||||
withdrawalManager = withdraw.NewManager(withdrawalCfg)
|
||||
withdrawalManager = withdraw.NewManager(withdrawalCfg, blockHeight)
|
||||
|
||||
// Static address loop-in manager setup.
|
||||
staticAddressLoopInStore := loopin.NewSqlStore(
|
||||
|
|
@ -631,7 +639,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
ValidateLoopInContract: loop.ValidateLoopInContract,
|
||||
MaxStaticAddrHtlcFeePercentage: d.cfg.MaxStaticAddrHtlcFeePercentage,
|
||||
MaxStaticAddrHtlcBackupFeePercentage: d.cfg.MaxStaticAddrHtlcBackupFeePercentage,
|
||||
})
|
||||
}, blockHeight)
|
||||
|
||||
var (
|
||||
reservationManager *reservation.Manager
|
||||
|
|
@ -674,7 +682,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
}
|
||||
|
||||
instantOutManager = instantout.NewInstantOutManager(
|
||||
instantOutConfig,
|
||||
instantOutConfig, int32(blockHeight),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -711,7 +719,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
// started yet, so if we clean that up now, nothing else needs
|
||||
// to be shut down at this point.
|
||||
if err := d.macaroonService.Stop(); err != nil {
|
||||
log.Errorf("Error shutting down macaroon service: %v",
|
||||
errorf("Error shutting down macaroon service: %v",
|
||||
err)
|
||||
}
|
||||
cleanupMacaroonStore()
|
||||
|
|
@ -728,7 +736,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Infof("Starting swap client")
|
||||
infof("Starting swap client")
|
||||
err := d.impl.Run(d.mainCtx, d.statusChan)
|
||||
if err != nil {
|
||||
// Notify the main error handler goroutine that
|
||||
|
|
@ -737,7 +745,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
// channel is sufficiently buffered.
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
log.Infof("Swap client stopped")
|
||||
infof("Swap client stopped")
|
||||
}()
|
||||
|
||||
// Start a goroutine that broadcasts swap updates to clients.
|
||||
|
|
@ -745,7 +753,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Infof("Waiting for updates")
|
||||
infof("Waiting for updates")
|
||||
d.processStatusUpdates(d.mainCtx)
|
||||
}()
|
||||
|
||||
|
|
@ -753,13 +761,13 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Info("Starting liquidity manager")
|
||||
infof("Starting liquidity manager")
|
||||
err := d.liquidityMgr.Run(d.mainCtx)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
|
||||
log.Info("Liquidity manager stopped")
|
||||
infof("Liquidity manager stopped")
|
||||
}()
|
||||
|
||||
// Start the reservation manager.
|
||||
|
|
@ -769,19 +777,11 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
// We need to know the current block height to properly
|
||||
// initialize the reservation manager.
|
||||
getInfo, err := d.lnd.Client.GetInfo(d.mainCtx)
|
||||
if err != nil {
|
||||
d.internalErrChan <- err
|
||||
return
|
||||
}
|
||||
infof("Starting reservation manager")
|
||||
defer infof("Reservation manager stopped")
|
||||
|
||||
log.Info("Starting reservation manager")
|
||||
defer log.Info("Reservation manager stopped")
|
||||
|
||||
err = d.reservationManager.Run(
|
||||
d.mainCtx, int32(getInfo.BlockHeight), initChan,
|
||||
err := d.reservationManager.Run(
|
||||
d.mainCtx, int32(blockHeight), initChan,
|
||||
)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
d.internalErrChan <- err
|
||||
|
|
@ -809,18 +809,10 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
getInfo, err := d.lnd.Client.GetInfo(d.mainCtx)
|
||||
if err != nil {
|
||||
d.internalErrChan <- err
|
||||
return
|
||||
}
|
||||
infof("Starting instantout manager")
|
||||
defer infof("Instantout manager stopped")
|
||||
|
||||
log.Info("Starting instantout manager")
|
||||
defer log.Info("Instantout manager stopped")
|
||||
|
||||
err = d.instantOutManager.Run(
|
||||
d.mainCtx, initChan, int32(getInfo.BlockHeight),
|
||||
)
|
||||
err := d.instantOutManager.Run(d.mainCtx, initChan)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
|
|
@ -846,12 +838,12 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
log.Info("Starting static address manager...")
|
||||
err = staticAddressManager.Run(d.mainCtx)
|
||||
infof("Starting static address manager...")
|
||||
err := staticAddressManager.Run(d.mainCtx)
|
||||
if err != nil && !errors.Is(context.Canceled, err) {
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
log.Info("Static address manager stopped")
|
||||
infof("Static address manager stopped")
|
||||
}()
|
||||
}
|
||||
|
||||
|
|
@ -861,20 +853,12 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
// Lnd's GetInfo call supplies us with the current block
|
||||
// height.
|
||||
info, err := d.lnd.Client.GetInfo(d.mainCtx)
|
||||
if err != nil {
|
||||
d.internalErrChan <- err
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Starting static address deposit manager...")
|
||||
err = depositManager.Run(d.mainCtx, info.BlockHeight)
|
||||
infof("Starting static address deposit manager...")
|
||||
err := depositManager.Run(d.mainCtx)
|
||||
if err != nil && !errors.Is(context.Canceled, err) {
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
log.Info("Static address deposit manager stopped")
|
||||
infof("Static address deposit manager stopped")
|
||||
}()
|
||||
depositManager.WaitInitComplete()
|
||||
}
|
||||
|
|
@ -885,21 +869,13 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
// Lnd's GetInfo call supplies us with the current block
|
||||
// height.
|
||||
info, err := d.lnd.Client.GetInfo(d.mainCtx)
|
||||
if err != nil {
|
||||
d.internalErrChan <- err
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Starting static address deposit withdrawal " +
|
||||
infof("Starting static address deposit withdrawal " +
|
||||
"manager...")
|
||||
err = withdrawalManager.Run(d.mainCtx, info.BlockHeight)
|
||||
err := withdrawalManager.Run(d.mainCtx)
|
||||
if err != nil && !errors.Is(context.Canceled, err) {
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
log.Info("Static address deposit withdrawal manager " +
|
||||
infof("Static address deposit withdrawal manager " +
|
||||
"stopped")
|
||||
}()
|
||||
withdrawalManager.WaitInitComplete()
|
||||
|
|
@ -911,23 +887,12 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
go func() {
|
||||
defer d.wg.Done()
|
||||
|
||||
// Lnd's GetInfo call supplies us with the current block
|
||||
// height.
|
||||
info, err := d.lnd.Client.GetInfo(d.mainCtx)
|
||||
if err != nil {
|
||||
d.internalErrChan <- err
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Starting static address loop-in manager...")
|
||||
err = staticLoopInManager.Run(
|
||||
d.mainCtx, info.BlockHeight,
|
||||
)
|
||||
infof("Starting static address loop-in manager...")
|
||||
err := staticLoopInManager.Run(d.mainCtx)
|
||||
if err != nil && !errors.Is(context.Canceled, err) {
|
||||
d.internalErrChan <- err
|
||||
}
|
||||
log.Info("Starting static address loop-in manager " +
|
||||
infof("Starting static address loop-in manager " +
|
||||
"stopped")
|
||||
}()
|
||||
staticLoopInManager.WaitInitComplete()
|
||||
|
|
@ -948,7 +913,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
// signal the caller that we're done.
|
||||
select {
|
||||
case runtimeErr = <-d.internalErrChan:
|
||||
log.Errorf("Runtime error in daemon, shutting down: "+
|
||||
errorf("Runtime error in daemon, shutting down: "+
|
||||
"%v", runtimeErr)
|
||||
|
||||
case <-d.quit:
|
||||
|
|
@ -958,7 +923,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
|
|||
// otherwise a caller might exit the process too early.
|
||||
d.stop()
|
||||
cleanupMacaroonStore()
|
||||
log.Info("Daemon exited")
|
||||
infof("Daemon exited")
|
||||
|
||||
// The caller expects exactly one message. So we send the error
|
||||
// even if it's nil because we cleanly shut down.
|
||||
|
|
@ -987,17 +952,17 @@ func (d *Daemon) stop() {
|
|||
|
||||
// As there is no swap activity anymore, we can forcefully shut down the
|
||||
// gRPC and HTTP servers now.
|
||||
log.Infof("Stopping gRPC server")
|
||||
infof("Stopping gRPC server")
|
||||
if d.grpcServer != nil {
|
||||
d.grpcServer.Stop()
|
||||
}
|
||||
log.Infof("Stopping REST server")
|
||||
infof("Stopping REST server")
|
||||
if d.restServer != nil {
|
||||
// Don't return the error here, we first want to give everything
|
||||
// else a chance to shut down cleanly.
|
||||
err := d.restServer.Close()
|
||||
if err != nil {
|
||||
log.Errorf("Error stopping REST server: %v", err)
|
||||
errorf("Error stopping REST server: %v", err)
|
||||
}
|
||||
}
|
||||
if d.restCtxCancel != nil {
|
||||
|
|
@ -1007,7 +972,7 @@ func (d *Daemon) stop() {
|
|||
if d.macaroonService != nil {
|
||||
err := d.macaroonService.Stop()
|
||||
if err != nil {
|
||||
log.Errorf("Error stopping macaroon service: %v", err)
|
||||
errorf("Error stopping macaroon service: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
40
loopd/log.go
40
loopd/log.go
|
|
@ -1,6 +1,8 @@
|
|||
package loopd
|
||||
|
||||
import (
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/btcsuite/btclog/v2"
|
||||
"github.com/lightninglabs/aperture/l402"
|
||||
"github.com/lightninglabs/lndclient"
|
||||
|
|
@ -22,18 +24,50 @@ import (
|
|||
const Subsystem = "LOOPD"
|
||||
|
||||
var (
|
||||
log btclog.Logger
|
||||
log_ atomic.Pointer[btclog.Logger]
|
||||
interceptor signal.Interceptor
|
||||
)
|
||||
|
||||
// log returns active logger.
|
||||
func log() btclog.Logger {
|
||||
return *log_.Load()
|
||||
}
|
||||
|
||||
// setLogger uses a specified Logger to output package logging info.
|
||||
func setLogger(logger btclog.Logger) {
|
||||
log_.Store(&logger)
|
||||
}
|
||||
|
||||
// tracef logs a message with level TRACE.
|
||||
func tracef(format string, params ...interface{}) {
|
||||
log().Tracef(format, params...)
|
||||
}
|
||||
|
||||
// infof logs a message with level INFO.
|
||||
func infof(format string, params ...interface{}) {
|
||||
log().Infof(format, params...)
|
||||
}
|
||||
|
||||
// warnf logs a message with level WARN.
|
||||
func warnf(format string, params ...interface{}) {
|
||||
log().Warnf(format, params...)
|
||||
}
|
||||
|
||||
// errorf logs a message with level ERROR.
|
||||
func errorf(format string, params ...interface{}) {
|
||||
log().Errorf(format, params...)
|
||||
}
|
||||
|
||||
// SetupLoggers initializes all package-global logger variables.
|
||||
func SetupLoggers(root *build.SubLoggerManager, intercept signal.Interceptor) {
|
||||
genLogger := genSubLogger(root, intercept)
|
||||
|
||||
log = build.NewSubLogger(Subsystem, genLogger)
|
||||
logger := build.NewSubLogger(Subsystem, genLogger)
|
||||
setLogger(logger)
|
||||
|
||||
interceptor = intercept
|
||||
|
||||
lnd.SetSubLogger(root, Subsystem, log)
|
||||
lnd.SetSubLogger(root, Subsystem, logger)
|
||||
lnd.AddSubLogger(root, "LOOP", intercept, loop.UseLogger)
|
||||
lnd.AddSubLogger(root, "SWEEP", intercept, sweepbatcher.UseLogger)
|
||||
lnd.AddSubLogger(root, "LNDC", intercept, lndclient.UseLogger)
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@ func needSqlMigration(cfg *Config) bool {
|
|||
// any deleted files occasionally (reboot, etc).
|
||||
sqliteDBPath := filepath.Join(cfg.DataDir, "loop_sqlite.db")
|
||||
if lnrpc.FileExists(sqliteDBPath) {
|
||||
log.Infof("Found sqlite db at %v, skipping migration",
|
||||
infof("Found sqlite db at %v, skipping migration",
|
||||
sqliteDBPath)
|
||||
|
||||
return false
|
||||
|
|
|
|||
|
|
@ -222,7 +222,7 @@ func Run(rpcCfg RPCConfig) error {
|
|||
}
|
||||
|
||||
// Print the version before executing either primary directive.
|
||||
log.Infof("Version: %v", loop.Version())
|
||||
infof("Version: %v", loop.Version())
|
||||
|
||||
lisCfg := NewListenerConfig(&config, rpcCfg)
|
||||
|
||||
|
|
@ -235,7 +235,7 @@ func Run(rpcCfg RPCConfig) error {
|
|||
|
||||
select {
|
||||
case <-interceptor.ShutdownChannel():
|
||||
log.Infof("Received SIGINT (Ctrl+C).")
|
||||
infof("Received SIGINT (Ctrl+C).")
|
||||
daemon.Stop()
|
||||
|
||||
// The above stop will return immediately. But we'll be
|
||||
|
|
|
|||
|
|
@ -112,7 +112,7 @@ func (s *swapClientServer) LoopOut(ctx context.Context,
|
|||
in *looprpc.LoopOutRequest) (
|
||||
*looprpc.SwapResponse, error) {
|
||||
|
||||
log.Infof("Loop out request received")
|
||||
infof("Loop out request received")
|
||||
|
||||
// Note that LoopOutRequest.PaymentTimeout is unsigned and therefore
|
||||
// cannot be negative.
|
||||
|
|
@ -257,7 +257,7 @@ func (s *swapClientServer) LoopOut(ctx context.Context,
|
|||
|
||||
info, err := s.impl.LoopOut(ctx, req)
|
||||
if err != nil {
|
||||
log.Errorf("LoopOut: %v", err)
|
||||
errorf("LoopOut: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
|
@ -461,7 +461,7 @@ func (s *swapClientServer) marshallSwap(ctx context.Context,
|
|||
func (s *swapClientServer) Monitor(in *looprpc.MonitorRequest,
|
||||
server looprpc.SwapClient_MonitorServer) error {
|
||||
|
||||
log.Infof("Monitor request received")
|
||||
infof("Monitor request received")
|
||||
|
||||
send := func(info loop.SwapInfo) error {
|
||||
rpcSwap, err := s.marshallSwap(server.Context(), &info)
|
||||
|
|
@ -732,11 +732,11 @@ func (s *swapClientServer) AbandonSwap(ctx context.Context,
|
|||
func (s *swapClientServer) LoopOutTerms(ctx context.Context,
|
||||
_ *looprpc.TermsRequest) (*looprpc.OutTermsResponse, error) {
|
||||
|
||||
log.Infof("Loop out terms request received")
|
||||
infof("Loop out terms request received")
|
||||
|
||||
terms, err := s.impl.LoopOutTerms(ctx, defaultLoopdInitiator)
|
||||
if err != nil {
|
||||
log.Errorf("Terms request: %v", err)
|
||||
errorf("Terms request: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
|
@ -822,11 +822,11 @@ func (s *swapClientServer) LoopOutQuote(ctx context.Context,
|
|||
func (s *swapClientServer) GetLoopInTerms(ctx context.Context,
|
||||
_ *looprpc.TermsRequest) (*looprpc.InTermsResponse, error) {
|
||||
|
||||
log.Infof("Loop in terms request received")
|
||||
infof("Loop in terms request received")
|
||||
|
||||
terms, err := s.impl.LoopInTerms(ctx, defaultLoopdInitiator)
|
||||
if err != nil {
|
||||
log.Errorf("Terms request: %v", err)
|
||||
errorf("Terms request: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
|
@ -840,7 +840,7 @@ func (s *swapClientServer) GetLoopInTerms(ctx context.Context,
|
|||
func (s *swapClientServer) GetLoopInQuote(ctx context.Context,
|
||||
req *looprpc.QuoteRequest) (*looprpc.InQuoteResponse, error) {
|
||||
|
||||
log.Infof("Loop in quote request received")
|
||||
infof("Loop in quote request received")
|
||||
|
||||
var (
|
||||
numDeposits = uint32(len(req.DepositOutpoints))
|
||||
|
|
@ -981,7 +981,7 @@ func unmarshallHopHint(rpcHint *swapserverrpc.HopHint) (zpay32.HopHint, error) {
|
|||
func (s *swapClientServer) Probe(ctx context.Context,
|
||||
req *looprpc.ProbeRequest) (*looprpc.ProbeResponse, error) {
|
||||
|
||||
log.Infof("Probe request received")
|
||||
infof("Probe request received")
|
||||
|
||||
var lastHop *route.Vertex
|
||||
if req.LastHop != nil {
|
||||
|
|
@ -1013,7 +1013,7 @@ func (s *swapClientServer) Probe(ctx context.Context,
|
|||
func (s *swapClientServer) LoopIn(ctx context.Context,
|
||||
in *looprpc.LoopInRequest) (*looprpc.SwapResponse, error) {
|
||||
|
||||
log.Infof("Loop in request received")
|
||||
infof("Loop in request received")
|
||||
|
||||
htlcConfTarget, err := validateLoopInRequest(
|
||||
in.HtlcConfTarget, in.ExternalHtlc, 0, in.Amt,
|
||||
|
|
@ -1052,7 +1052,7 @@ func (s *swapClientServer) LoopIn(ctx context.Context,
|
|||
}
|
||||
swapInfo, err := s.impl.LoopIn(ctx, req)
|
||||
if err != nil {
|
||||
log.Errorf("Loop in: %v", err)
|
||||
errorf("Loop in: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
|
@ -1079,7 +1079,7 @@ func (s *swapClientServer) LoopIn(ctx context.Context,
|
|||
func (s *swapClientServer) GetL402Tokens(ctx context.Context,
|
||||
_ *looprpc.TokensRequest) (*looprpc.TokensResponse, error) {
|
||||
|
||||
log.Infof("Get L402 tokens request received")
|
||||
infof("Get L402 tokens request received")
|
||||
|
||||
tokens, err := s.impl.L402Store.AllTokens()
|
||||
if err != nil {
|
||||
|
|
@ -1128,7 +1128,7 @@ func (s *swapClientServer) GetL402Tokens(ctx context.Context,
|
|||
func (s *swapClientServer) GetLsatTokens(ctx context.Context,
|
||||
req *looprpc.TokensRequest) (*looprpc.TokensResponse, error) {
|
||||
|
||||
log.Warnf("Received deprecated call GetLsatTokens. Please update the " +
|
||||
warnf("Received deprecated call GetLsatTokens. Please update the " +
|
||||
"client software. Calling GetL402Tokens now.")
|
||||
|
||||
return s.GetL402Tokens(ctx, req)
|
||||
|
|
@ -1754,7 +1754,7 @@ func (s *swapClientServer) StaticAddressLoopIn(ctx context.Context,
|
|||
in *looprpc.StaticAddressLoopInRequest) (
|
||||
*looprpc.StaticAddressLoopInResponse, error) {
|
||||
|
||||
log.Infof("Static loop-in request received")
|
||||
infof("Static loop-in request received")
|
||||
|
||||
routeHints, err := unmarshallRouteHints(in.RouteHints)
|
||||
if err != nil {
|
||||
|
|
@ -2187,52 +2187,52 @@ func validateLoopOutRequest(ctx context.Context, lnd lndclient.LightningClient,
|
|||
func hasBandwidth(channels []lndclient.ChannelInfo, amt btcutil.Amount,
|
||||
maxParts int) (bool, int) {
|
||||
|
||||
log.Tracef("Checking if %v sats can be routed with %v parts over "+
|
||||
tracef("Checking if %v sats can be routed with %v parts over "+
|
||||
"channel set of length %v", amt, maxParts, len(channels))
|
||||
|
||||
localBalances := make([]btcutil.Amount, len(channels))
|
||||
var totalBandwidth btcutil.Amount
|
||||
for i, channel := range channels {
|
||||
log.Tracef("Channel %v: local=%v remote=%v", channel.ChannelID,
|
||||
tracef("Channel %v: local=%v remote=%v", channel.ChannelID,
|
||||
channel.LocalBalance, channel.RemoteBalance)
|
||||
|
||||
localBalances[i] = channel.LocalBalance
|
||||
totalBandwidth += channel.LocalBalance
|
||||
}
|
||||
|
||||
log.Tracef("Total bandwidth: %v", totalBandwidth)
|
||||
tracef("Total bandwidth: %v", totalBandwidth)
|
||||
if totalBandwidth < amt {
|
||||
return false, 0
|
||||
}
|
||||
|
||||
logLocalBalances := func(shard int) {
|
||||
log.Tracef("Local balances for %v shards:", shard)
|
||||
tracef("Local balances for %v shards:", shard)
|
||||
for i, balance := range localBalances {
|
||||
log.Tracef("Channel %v: localBalances[%v]=%v",
|
||||
tracef("Channel %v: localBalances[%v]=%v",
|
||||
channels[i].ChannelID, i, balance)
|
||||
}
|
||||
}
|
||||
|
||||
split := amt
|
||||
for shard := 0; shard <= maxParts; {
|
||||
log.Tracef("Trying to split %v sats into %v parts", amt, shard)
|
||||
tracef("Trying to split %v sats into %v parts", amt, shard)
|
||||
|
||||
paid := false
|
||||
for i := 0; i < len(localBalances); i++ {
|
||||
// TODO(hieblmi): Consider channel reserves because the
|
||||
// channel can't send its full local balance.
|
||||
if localBalances[i] >= split {
|
||||
log.Tracef("len(shards)=%v: Local channel "+
|
||||
tracef("len(shards)=%v: Local channel "+
|
||||
"balance %v can pay %v sats",
|
||||
shard, localBalances[i], split)
|
||||
|
||||
localBalances[i] -= split
|
||||
log.Tracef("len(shards)=%v: Subtracted "+
|
||||
tracef("len(shards)=%v: Subtracted "+
|
||||
"%v sats from localBalance[%v]=%v",
|
||||
shard, split, i, localBalances[i])
|
||||
|
||||
amt -= split
|
||||
log.Tracef("len(shards)=%v: Remaining total "+
|
||||
tracef("len(shards)=%v: Remaining total "+
|
||||
"amount amt=%v", shard, amt)
|
||||
|
||||
paid = true
|
||||
|
|
@ -2245,26 +2245,26 @@ func hasBandwidth(channels []lndclient.ChannelInfo, amt btcutil.Amount,
|
|||
logLocalBalances(shard)
|
||||
|
||||
if amt == 0 {
|
||||
log.Tracef("Payment is routable with %v part(s)", shard)
|
||||
tracef("Payment is routable with %v part(s)", shard)
|
||||
|
||||
return true, shard
|
||||
}
|
||||
|
||||
if !paid {
|
||||
log.Tracef("len(shards)=%v: No channel could pay %v "+
|
||||
tracef("len(shards)=%v: No channel could pay %v "+
|
||||
"sats, halving payment to %v and trying again",
|
||||
split/2)
|
||||
|
||||
split /= 2
|
||||
} else {
|
||||
log.Tracef("len(shards)=%v: Payment was made, trying "+
|
||||
tracef("len(shards)=%v: Payment was made, trying "+
|
||||
"to pay remaining sats %v", shard, amt)
|
||||
|
||||
split = amt
|
||||
}
|
||||
}
|
||||
|
||||
log.Tracef("Payment is not routable, remaining amount that can't be "+
|
||||
tracef("Payment is not routable, remaining amount that can't be "+
|
||||
"sent: %v sats", amt)
|
||||
|
||||
logLocalBalances(maxParts)
|
||||
|
|
|
|||
|
|
@ -491,7 +491,8 @@ func TestValidateLoopOutRequest(t *testing.T) {
|
|||
logger := btclog.NewSLogger(
|
||||
btclog.NewDefaultHandler(os.Stdout),
|
||||
)
|
||||
log = logger.SubSystem(Subsystem)
|
||||
setLogger(logger.SubSystem(Subsystem))
|
||||
|
||||
conf, err := validateLoopOutRequest(
|
||||
ctx, lnd.Client, &test.chain, req,
|
||||
test.destAddr, test.maxParts,
|
||||
|
|
|
|||
|
|
@ -57,12 +57,12 @@ func getClient(cfg *Config, swapDb loopdb.SwapStore,
|
|||
}
|
||||
|
||||
if cfg.MaxL402Cost == defaultCost && cfg.MaxLSATCost != 0 {
|
||||
log.Warnf("Option maxlsatcost is deprecated and will be " +
|
||||
warnf("Option maxlsatcost is deprecated and will be " +
|
||||
"removed. Switch to maxl402cost.")
|
||||
clientConfig.MaxL402Cost = btcutil.Amount(cfg.MaxLSATCost)
|
||||
}
|
||||
if cfg.MaxL402Fee == defaultFee && cfg.MaxLSATFee != 0 {
|
||||
log.Warnf("Option maxlsatfee is deprecated and will be " +
|
||||
warnf("Option maxlsatfee is deprecated and will be " +
|
||||
"removed. Switch to maxl402fee.")
|
||||
clientConfig.MaxL402Fee = btcutil.Amount(cfg.MaxLSATFee)
|
||||
}
|
||||
|
|
@ -87,7 +87,7 @@ func openDatabase(cfg *Config, chainParams *chaincfg.Params) (loopdb.SwapStore,
|
|||
)
|
||||
switch cfg.DatabaseBackend {
|
||||
case DatabaseBackendSqlite:
|
||||
log.Infof("Opening sqlite3 database at: %v",
|
||||
infof("Opening sqlite3 database at: %v",
|
||||
cfg.Sqlite.DatabaseFileName)
|
||||
|
||||
db, err = loopdb.NewSqliteStore(cfg.Sqlite, chainParams)
|
||||
|
|
@ -97,7 +97,7 @@ func openDatabase(cfg *Config, chainParams *chaincfg.Params) (loopdb.SwapStore,
|
|||
baseDb = *db.(*loopdb.SqliteSwapStore).BaseDB
|
||||
|
||||
case DatabaseBackendPostgres:
|
||||
log.Infof("Opening postgres database at: %v",
|
||||
infof("Opening postgres database at: %v",
|
||||
cfg.Postgres.DSN(true))
|
||||
|
||||
db, err = loopdb.NewPostgresStore(cfg.Postgres, chainParams)
|
||||
|
|
|
|||
|
|
@ -6,9 +6,9 @@ import (
|
|||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/bbolt"
|
||||
"github.com/lightninglabs/loop/labels"
|
||||
"github.com/lightningnetwork/lnd/routing/route"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// LoopInContract contains the data that is serialized to persistent storage for
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import (
|
|||
"fmt"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/coreos/bbolt"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import (
|
|||
"fmt"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/coreos/bbolt"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// noMigrationAvailable is the fall back migration in case there is no migration
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import (
|
|||
"fmt"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/coreos/bbolt"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// migrateSwapPublicationDeadline migrates the database to v02, by adding the
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import (
|
|||
"fmt"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/coreos/bbolt"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// migrateLastHop migrates the database to v03, replacing the never used loop in
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import (
|
|||
"fmt"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/coreos/bbolt"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// migrateUpdates migrates the swap updates to add an additional level of
|
||||
|
|
|
|||
|
|
@ -8,8 +8,8 @@ import (
|
|||
"testing"
|
||||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/coreos/bbolt"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// TestMigrationUpdates asserts that the swap updates migration is carried out
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import (
|
|||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/coreos/bbolt"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
// DumpDB dumps go code describing the contents of the database to stdout. This
|
||||
|
|
|
|||
|
|
@ -13,8 +13,8 @@ import (
|
|||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/coreos/bbolt"
|
||||
"github.com/lightningnetwork/lnd/lntypes"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
|
|||
|
|
@ -11,12 +11,12 @@ import (
|
|||
|
||||
"github.com/btcsuite/btcd/chaincfg"
|
||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||
"github.com/coreos/bbolt"
|
||||
"github.com/lightninglabs/loop/test"
|
||||
"github.com/lightningnetwork/lnd/keychain"
|
||||
"github.com/lightningnetwork/lnd/lntypes"
|
||||
"github.com/lightningnetwork/lnd/routing/route"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.etcd.io/bbolt"
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
|
|||
|
|
@ -873,7 +873,7 @@ func (s *loopOutSwap) payInvoiceAsync(ctx context.Context,
|
|||
}
|
||||
|
||||
if err := s.swapKit.server.ReportRoutingResult(
|
||||
ctx, s.swapInfo().SwapHash, s.swapInvoicePaymentAddr,
|
||||
ctx, s.hash, s.swapInvoicePaymentAddr,
|
||||
reportType, paymentSuccess, int32(attempts),
|
||||
dt.Milliseconds(),
|
||||
); err != nil {
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/btcsuite/btcd/btcec/v2"
|
||||
"github.com/btcsuite/btcd/btcec/v2/schnorr"
|
||||
|
|
@ -52,14 +53,17 @@ type Manager struct {
|
|||
|
||||
sync.Mutex
|
||||
|
||||
currentHeight int32
|
||||
currentHeight atomic.Int32
|
||||
}
|
||||
|
||||
// NewManager creates a new address manager.
|
||||
func NewManager(cfg *ManagerConfig) *Manager {
|
||||
return &Manager{
|
||||
func NewManager(cfg *ManagerConfig, currentHeight int32) *Manager {
|
||||
m := &Manager{
|
||||
cfg: cfg,
|
||||
}
|
||||
m.currentHeight.Store(currentHeight)
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
// Run runs the address manager.
|
||||
|
|
@ -74,7 +78,7 @@ func (m *Manager) Run(ctx context.Context) error {
|
|||
for {
|
||||
select {
|
||||
case currentHeight := <-newBlockChan:
|
||||
m.currentHeight = currentHeight
|
||||
m.currentHeight.Store(currentHeight)
|
||||
|
||||
case err = <-newBlockErrChan:
|
||||
return err
|
||||
|
|
@ -110,11 +114,6 @@ func (m *Manager) NewAddress(ctx context.Context) (*btcutil.AddressTaproot,
|
|||
}
|
||||
m.Unlock()
|
||||
|
||||
// Ensure that we have that we have a sane current block height.
|
||||
if m.currentHeight == 0 {
|
||||
return nil, fmt.Errorf("current block height is unknown")
|
||||
}
|
||||
|
||||
// We are fetching a new L402 token from the server. There is one static
|
||||
// address per L402 token allowed.
|
||||
err = m.cfg.FetchL402(ctx)
|
||||
|
|
@ -176,7 +175,7 @@ func (m *Manager) NewAddress(ctx context.Context) (*btcutil.AddressTaproot,
|
|||
ProtocolVersion: version.AddressProtocolVersion(
|
||||
protocolVersion,
|
||||
),
|
||||
InitiationHeight: m.currentHeight,
|
||||
InitiationHeight: m.currentHeight.Load(),
|
||||
}
|
||||
err = m.cfg.Store.CreateStaticAddress(ctx, addrParams)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -154,6 +154,9 @@ type ManagerTestContext struct {
|
|||
// NewAddressManagerTestContext creates a new test context for the static
|
||||
// address manager.
|
||||
func NewAddressManagerTestContext(t *testing.T) *ManagerTestContext {
|
||||
ctxb, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
mockLnd := test.NewMockLnd()
|
||||
lndContext := test.NewContext(t, mockLnd)
|
||||
|
||||
|
|
@ -183,7 +186,10 @@ func NewAddressManagerTestContext(t *testing.T) *ManagerTestContext {
|
|||
FetchL402: func(context.Context) error { return nil },
|
||||
}
|
||||
|
||||
manager := NewManager(cfg)
|
||||
getInfo, err := mockLnd.Client.GetInfo(ctxb)
|
||||
require.NoError(t, err)
|
||||
|
||||
manager := NewManager(cfg, int32(getInfo.BlockHeight))
|
||||
|
||||
return &ManagerTestContext{
|
||||
manager: manager,
|
||||
|
|
|
|||
|
|
@ -84,9 +84,6 @@ type Manager struct {
|
|||
// activeDeposits contains all the active static address outputs.
|
||||
activeDeposits map[wire.OutPoint]*FSM
|
||||
|
||||
// initiationHeight stores the currently best known block height.
|
||||
initiationHeight uint32
|
||||
|
||||
// deposits contains all the deposits that have ever been made to the
|
||||
// static address. This field is used to store and recover deposits. It
|
||||
// also serves as basis for reconciliation of newly detected deposits by
|
||||
|
|
@ -111,9 +108,7 @@ func NewManager(cfg *ManagerConfig) *Manager {
|
|||
}
|
||||
|
||||
// Run runs the address manager.
|
||||
func (m *Manager) Run(ctx context.Context, currentHeight uint32) error {
|
||||
m.initiationHeight = currentHeight
|
||||
|
||||
func (m *Manager) Run(ctx context.Context) error {
|
||||
newBlockChan, newBlockErrChan, err := m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx) //nolint:lll
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
|
|||
|
|
@ -219,10 +219,7 @@ func TestManager(t *testing.T) {
|
|||
|
||||
// Start the deposit manager.
|
||||
go func() {
|
||||
err := testContext.manager.Run(
|
||||
ctx, uint32(testContext.mockLnd.Height),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, testContext.manager.Run(ctx))
|
||||
}()
|
||||
|
||||
// Ensure that the manager has been initialized.
|
||||
|
|
|
|||
|
|
@ -139,8 +139,8 @@ type Manager struct {
|
|||
}
|
||||
|
||||
// NewManager creates a new deposit withdrawal manager.
|
||||
func NewManager(cfg *Config) *Manager {
|
||||
return &Manager{
|
||||
func NewManager(cfg *Config, currentHeight uint32) *Manager {
|
||||
m := &Manager{
|
||||
cfg: cfg,
|
||||
initChan: make(chan struct{}),
|
||||
newLoopInChan: make(chan *newSwapRequest),
|
||||
|
|
@ -148,12 +148,13 @@ func NewManager(cfg *Config) *Manager {
|
|||
errChan: make(chan error),
|
||||
activeLoopIns: make(map[lntypes.Hash]*FSM),
|
||||
}
|
||||
m.currentHeight.Store(currentHeight)
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
// Run runs the static address loop-in manager.
|
||||
func (m *Manager) Run(ctx context.Context, currentHeight uint32) error {
|
||||
m.currentHeight.Store(currentHeight)
|
||||
|
||||
func (m *Manager) Run(ctx context.Context) error {
|
||||
registerBlockNtfn := m.cfg.ChainNotifier.RegisterBlockEpochNtfn
|
||||
newBlockChan, newBlockErrChan, err := registerBlockNtfn(ctx)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import (
|
|||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/btcsuite/btcd/btcec/v2/schnorr"
|
||||
"github.com/btcsuite/btcd/btcec/v2/schnorr/musig2"
|
||||
|
|
@ -106,7 +107,7 @@ type Manager struct {
|
|||
errChan chan error
|
||||
|
||||
// initiationHeight stores the currently best known block height.
|
||||
initiationHeight uint32
|
||||
initiationHeight atomic.Uint32
|
||||
|
||||
// finalizedWithdrawalTx are the finalized withdrawal transactions that
|
||||
// are published to the network and re-published on block arrivals.
|
||||
|
|
@ -114,8 +115,8 @@ type Manager struct {
|
|||
}
|
||||
|
||||
// NewManager creates a new deposit withdrawal manager.
|
||||
func NewManager(cfg *ManagerConfig) *Manager {
|
||||
return &Manager{
|
||||
func NewManager(cfg *ManagerConfig, currentHeight uint32) *Manager {
|
||||
m := &Manager{
|
||||
cfg: cfg,
|
||||
initChan: make(chan struct{}),
|
||||
finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx),
|
||||
|
|
@ -123,12 +124,13 @@ func NewManager(cfg *ManagerConfig) *Manager {
|
|||
newWithdrawalRequestChan: make(chan newWithdrawalRequest),
|
||||
errChan: make(chan error),
|
||||
}
|
||||
m.initiationHeight.Store(currentHeight)
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
// Run runs the deposit withdrawal manager.
|
||||
func (m *Manager) Run(ctx context.Context, currentHeight uint32) error {
|
||||
m.initiationHeight = currentHeight
|
||||
|
||||
func (m *Manager) Run(ctx context.Context) error {
|
||||
newBlockChan, newBlockErrChan, err :=
|
||||
m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx)
|
||||
|
||||
|
|
@ -479,7 +481,7 @@ func (m *Manager) handleWithdrawal(ctx context.Context,
|
|||
|
||||
confChan, errChan, err := m.cfg.ChainNotifier.RegisterConfirmationsNtfn(
|
||||
ctx, &txHash, withdrawalPkScript, MinConfs,
|
||||
int32(m.initiationHeight),
|
||||
int32(m.initiationHeight.Load()),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue