cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Plumb contexts through

Browse Source
This commit is contained in:
Aayush Rajasekaran
2021-12-11 17:04:00 -05:00
parent 6c31cecc7d
commit dfb65ed89f
71 changed files with 595 additions and 577 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paychmgr/accessorcache.go
+7 -3
View File
@@ -1,6 +1,10 @@
package paychmgr
import "github.com/filecoin-project/go-address"
import (
"context"
"github.com/filecoin-project/go-address"
)
// accessorByFromTo gets a channel accessor for a given from / to pair.
// The channel accessor facilitates locking a channel so that operations
@@ -36,10 +40,10 @@ func (pm *Manager) accessorByFromTo(from address.Address, to address.Address) (*
// The channel accessor facilitates locking a channel so that operations
// must be performed sequentially on a channel (but can be performed at
// the same time on different channels).
func (pm *Manager) accessorByAddress(ch address.Address) (*channelAccessor, error) {
func (pm *Manager) accessorByAddress(ctx context.Context, ch address.Address) (*channelAccessor, error) {
// Get the channel from / to
pm.lk.RLock()
channelInfo, err := pm.store.ByAddress(ch)
channelInfo, err := pm.store.ByAddress(ctx, ch)
pm.lk.RUnlock()
if err != nil {
return nil, err
paychmgr/manager.go
+26 -26
View File
@@ -92,7 +92,7 @@ func newManager(pchstore *Store, pchapi managerAPI) (*Manager, error) {
// Start restarts tracking of any messages that were sent to chain.
func (pm *Manager) Start() error {
return pm.restartPending()
return pm.restartPending(pm.ctx)
}
// Stop shuts down any processes used by the manager
@@ -110,27 +110,27 @@ func (pm *Manager) GetPaych(ctx context.Context, from, to address.Address, amt t
return chanAccessor.getPaych(ctx, amt)
}
func (pm *Manager) AvailableFunds(ch address.Address) (*api.ChannelAvailableFunds, error) {
ca, err := pm.accessorByAddress(ch)
func (pm *Manager) AvailableFunds(ctx context.Context, ch address.Address) (*api.ChannelAvailableFunds, error) {
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return nil, err
}
ci, err := ca.getChannelInfo(ch)
ci, err := ca.getChannelInfo(ctx, ch)
if err != nil {
return nil, err
}
return ca.availableFunds(ci.ChannelID)
return ca.availableFunds(ctx, ci.ChannelID)
}
func (pm *Manager) AvailableFundsByFromTo(from address.Address, to address.Address) (*api.ChannelAvailableFunds, error) {
func (pm *Manager) AvailableFundsByFromTo(ctx context.Context, from address.Address, to address.Address) (*api.ChannelAvailableFunds, error) {
ca, err := pm.accessorByFromTo(from, to)
if err != nil {
return nil, err
}
ci, err := ca.outboundActiveByFromTo(from, to)
ci, err := ca.outboundActiveByFromTo(ctx, from, to)
if err == ErrChannelNotTracked {
// If there is no active channel between from / to we still want to
// return an empty ChannelAvailableFunds, so that clients can check
@@ -151,7 +151,7 @@ func (pm *Manager) AvailableFundsByFromTo(from address.Address, to address.Addre
return nil, err
}
return ca.availableFunds(ci.ChannelID)
return ca.availableFunds(ctx, ci.ChannelID)
}
// GetPaychWaitReady waits until the create channel / add funds message with the
@@ -160,7 +160,7 @@ func (pm *Manager) AvailableFundsByFromTo(from address.Address, to address.Addre
func (pm *Manager) GetPaychWaitReady(ctx context.Context, mcid cid.Cid) (address.Address, error) {
// Find the channel associated with the message CID
pm.lk.Lock()
ci, err := pm.store.ByMessageCid(mcid)
ci, err := pm.store.ByMessageCid(ctx, mcid)
pm.lk.Unlock()
if err != nil {
@@ -178,25 +178,25 @@ func (pm *Manager) GetPaychWaitReady(ctx context.Context, mcid cid.Cid) (address
return chanAccessor.getPaychWaitReady(ctx, mcid)
}
func (pm *Manager) ListChannels() ([]address.Address, error) {
func (pm *Manager) ListChannels(ctx context.Context) ([]address.Address, error) {
// Need to take an exclusive lock here so that channel operations can't run
// in parallel (see channelLock)
pm.lk.Lock()
defer pm.lk.Unlock()
return pm.store.ListChannels()
return pm.store.ListChannels(ctx)
}
func (pm *Manager) GetChannelInfo(addr address.Address) (*ChannelInfo, error) {
ca, err := pm.accessorByAddress(addr)
func (pm *Manager) GetChannelInfo(ctx context.Context, addr address.Address) (*ChannelInfo, error) {
ca, err := pm.accessorByAddress(ctx, addr)
if err != nil {
return nil, err
}
return ca.getChannelInfo(addr)
return ca.getChannelInfo(ctx, addr)
}
func (pm *Manager) CreateVoucher(ctx context.Context, ch address.Address, voucher paych.SignedVoucher) (*api.VoucherCreateResult, error) {
ca, err := pm.accessorByAddress(ch)
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return nil, err
}
@@ -223,7 +223,7 @@ func (pm *Manager) CheckVoucherSpendable(ctx context.Context, ch address.Address
if len(proof) > 0 {
return false, errProofNotSupported
}
ca, err := pm.accessorByAddress(ch)
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return false, err
}
@@ -237,7 +237,7 @@ func (pm *Manager) AddVoucherOutbound(ctx context.Context, ch address.Address, s
if len(proof) > 0 {
return types.NewInt(0), errProofNotSupported
}
ca, err := pm.accessorByAddress(ch)
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return types.NewInt(0), err
}
@@ -283,7 +283,7 @@ func (pm *Manager) trackInboundChannel(ctx context.Context, ch address.Address)
defer pm.lk.Unlock()
// Check if channel is in store
ci, err := pm.store.ByAddress(ch)
ci, err := pm.store.ByAddress(ctx, ch)
if err == nil {
// Channel is in store, so it's already being tracked
return ci, nil
@@ -316,7 +316,7 @@ func (pm *Manager) trackInboundChannel(ctx context.Context, ch address.Address)
}
// Save channel to store
return pm.store.TrackChannel(stateCi)
return pm.store.TrackChannel(ctx, stateCi)
}
// TODO: secret vs proof doesn't make sense, there is only one, not two
@@ -324,23 +324,23 @@ func (pm *Manager) SubmitVoucher(ctx context.Context, ch address.Address, sv *pa
if len(proof) > 0 {
return cid.Undef, errProofNotSupported
}
ca, err := pm.accessorByAddress(ch)
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return cid.Undef, err
}
return ca.submitVoucher(ctx, ch, sv, secret)
}
func (pm *Manager) AllocateLane(ch address.Address) (uint64, error) {
ca, err := pm.accessorByAddress(ch)
func (pm *Manager) AllocateLane(ctx context.Context, ch address.Address) (uint64, error) {
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return 0, err
}
return ca.allocateLane(ch)
return ca.allocateLane(ctx, ch)
}
func (pm *Manager) ListVouchers(ctx context.Context, ch address.Address) ([]*VoucherInfo, error) {
ca, err := pm.accessorByAddress(ch)
ca, err := pm.accessorByAddress(ctx, ch)
if err != nil {
return nil, err
}
@@ -348,7 +348,7 @@ func (pm *Manager) ListVouchers(ctx context.Context, ch address.Address) ([]*Vou
}
func (pm *Manager) Settle(ctx context.Context, addr address.Address) (cid.Cid, error) {
ca, err := pm.accessorByAddress(addr)
ca, err := pm.accessorByAddress(ctx, addr)
if err != nil {
return cid.Undef, err
}
@@ -356,7 +356,7 @@ func (pm *Manager) Settle(ctx context.Context, addr address.Address) (cid.Cid, e
}
func (pm *Manager) Collect(ctx context.Context, addr address.Address) (cid.Cid, error) {
ca, err := pm.accessorByAddress(addr)
ca, err := pm.accessorByAddress(ctx, addr)
if err != nil {
return cid.Undef, err
}
paychmgr/paych.go
+19 -19
View File
@@ -95,18 +95,18 @@ func (ca *channelAccessor) messageBuilder(ctx context.Context, from address.Addr
return paych.Message(av, from), nil
}
func (ca *channelAccessor) getChannelInfo(addr address.Address) (*ChannelInfo, error) {
func (ca *channelAccessor) getChannelInfo(ctx context.Context, addr address.Address) (*ChannelInfo, error) {
ca.lk.Lock()
defer ca.lk.Unlock()
return ca.store.ByAddress(addr)
return ca.store.ByAddress(ctx, addr)
}
func (ca *channelAccessor) outboundActiveByFromTo(from, to address.Address) (*ChannelInfo, error) {
func (ca *channelAccessor) outboundActiveByFromTo(ctx context.Context, from, to address.Address) (*ChannelInfo, error) {
ca.lk.Lock()
defer ca.lk.Unlock()
return ca.store.OutboundActiveByFromTo(from, to)
return ca.store.OutboundActiveByFromTo(ctx, from, to)
}
// createVoucher creates a voucher with the given specification, setting its
@@ -118,7 +118,7 @@ func (ca *channelAccessor) createVoucher(ctx context.Context, ch address.Address
defer ca.lk.Unlock()
// Find the channel for the voucher
ci, err := ca.store.ByAddress(ch)
ci, err := ca.store.ByAddress(ctx, ch)
if err != nil {
return nil, xerrors.Errorf("failed to get channel info by address: %w", err)
}
@@ -229,7 +229,7 @@ func (ca *channelAccessor) checkVoucherValidUnlocked(ctx context.Context, ch add
}
// Check the voucher against the highest known voucher nonce / value
laneStates, err := ca.laneState(pchState, ch)
laneStates, err := ca.laneState(ctx, pchState, ch)
if err != nil {
return nil, err
}
@@ -298,7 +298,7 @@ func (ca *channelAccessor) checkVoucherSpendable(ctx context.Context, ch address
return false, err
}
ci, err := ca.store.ByAddress(ch)
ci, err := ca.store.ByAddress(ctx, ch)
if err != nil {
return false, err
}
@@ -351,7 +351,7 @@ func (ca *channelAccessor) addVoucher(ctx context.Context, ch address.Address, s
}
func (ca *channelAccessor) addVoucherUnlocked(ctx context.Context, ch address.Address, sv *paych.SignedVoucher, minDelta types.BigInt) (types.BigInt, error) {
ci, err := ca.store.ByAddress(ch)
ci, err := ca.store.ByAddress(ctx, ch)
if err != nil {
return types.BigInt{}, err
}
@@ -400,14 +400,14 @@ func (ca *channelAccessor) addVoucherUnlocked(ctx context.Context, ch address.Ad
ci.NextLane = sv.Lane + 1
}
return delta, ca.store.putChannelInfo(ci)
return delta, ca.store.putChannelInfo(ctx, ci)
}
func (ca *channelAccessor) submitVoucher(ctx context.Context, ch address.Address, sv *paych.SignedVoucher, secret []byte) (cid.Cid, error) {
ca.lk.Lock()
defer ca.lk.Unlock()
ci, err := ca.store.ByAddress(ch)
ci, err := ca.store.ByAddress(ctx, ch)
if err != nil {
return cid.Undef, err
}
@@ -453,7 +453,7 @@ func (ca *channelAccessor) submitVoucher(ctx context.Context, ch address.Address
}
// Mark the voucher and any lower-nonce vouchers as having been submitted
err = ca.store.MarkVoucherSubmitted(ci, sv)
err = ca.store.MarkVoucherSubmitted(ctx, ci, sv)
if err != nil {
return cid.Undef, err
}
@@ -461,11 +461,11 @@ func (ca *channelAccessor) submitVoucher(ctx context.Context, ch address.Address
return smsg.Cid(), nil
}
func (ca *channelAccessor) allocateLane(ch address.Address) (uint64, error) {
func (ca *channelAccessor) allocateLane(ctx context.Context, ch address.Address) (uint64, error) {
ca.lk.Lock()
defer ca.lk.Unlock()
return ca.store.AllocateLane(ch)
return ca.store.AllocateLane(ctx, ch)
}
func (ca *channelAccessor) listVouchers(ctx context.Context, ch address.Address) ([]*VoucherInfo, error) {
@@ -474,12 +474,12 @@ func (ca *channelAccessor) listVouchers(ctx context.Context, ch address.Address)
// TODO: just having a passthrough method like this feels odd. Seems like
// there should be some filtering we're doing here
return ca.store.VouchersForPaych(ch)
return ca.store.VouchersForPaych(ctx, ch)
}
// laneState gets the LaneStates from chain, then applies all vouchers in
// the data store over the chain state
func (ca *channelAccessor) laneState(state paych.State, ch address.Address) (map[uint64]paych.LaneState, error) {
func (ca *channelAccessor) laneState(ctx context.Context, state paych.State, ch address.Address) (map[uint64]paych.LaneState, error) {
// TODO: we probably want to call UpdateChannelState with all vouchers to be fully correct
// (but technically dont't need to)
@@ -501,7 +501,7 @@ func (ca *channelAccessor) laneState(state paych.State, ch address.Address) (map
}
// Apply locally stored vouchers
vouchers, err := ca.store.VouchersForPaych(ch)
vouchers, err := ca.store.VouchersForPaych(ctx, ch)
if err != nil && err != ErrChannelNotTracked {
return nil, err
}
@@ -583,7 +583,7 @@ func (ca *channelAccessor) settle(ctx context.Context, ch address.Address) (cid.
ca.lk.Lock()
defer ca.lk.Unlock()
ci, err := ca.store.ByAddress(ch)
ci, err := ca.store.ByAddress(ctx, ch)
if err != nil {
return cid.Undef, err
}
@@ -602,7 +602,7 @@ func (ca *channelAccessor) settle(ctx context.Context, ch address.Address) (cid.
}
ci.Settling = true
err = ca.store.putChannelInfo(ci)
err = ca.store.putChannelInfo(ctx, ci)
if err != nil {
log.Errorf("Error marking channel as settled: %s", err)
}
@@ -614,7 +614,7 @@ func (ca *channelAccessor) collect(ctx context.Context, ch address.Address) (cid
ca.lk.Lock()
defer ca.lk.Unlock()
ci, err := ca.store.ByAddress(ch)
ci, err := ca.store.ByAddress(ctx, ch)
if err != nil {
return cid.Undef, err
}
paychmgr/simple.go
+45 -45
View File
@@ -159,7 +159,7 @@ func (m *mergedFundsReq) sum() types.BigInt {
func (ca *channelAccessor) getPaych(ctx context.Context, amt types.BigInt) (address.Address, cid.Cid, error) {
// Add the request to add funds to a queue and wait for the result
freq := newFundsReq(ctx, amt)
ca.enqueue(freq)
ca.enqueue(ctx, freq)
select {
case res := <-freq.promise:
return res.channel, res.mcid, res.err
@@ -170,16 +170,16 @@ func (ca *channelAccessor) getPaych(ctx context.Context, amt types.BigInt) (addr
}
// Queue up an add funds operation
func (ca *channelAccessor) enqueue(task *fundsReq) {
func (ca *channelAccessor) enqueue(ctx context.Context, task *fundsReq) {
ca.lk.Lock()
defer ca.lk.Unlock()
ca.fundsReqQueue = append(ca.fundsReqQueue, task)
go ca.processQueue("") // nolint: errcheck
go ca.processQueue(ctx, "") // nolint: errcheck
}
// Run the operations in the queue
func (ca *channelAccessor) processQueue(channelID string) (*api.ChannelAvailableFunds, error) {
func (ca *channelAccessor) processQueue(ctx context.Context, channelID string) (*api.ChannelAvailableFunds, error) {
ca.lk.Lock()
defer ca.lk.Unlock()
@@ -188,7 +188,7 @@ func (ca *channelAccessor) processQueue(channelID string) (*api.ChannelAvailable
// If there's nothing in the queue, bail out
if len(ca.fundsReqQueue) == 0 {
return ca.currentAvailableFunds(channelID, types.NewInt(0))
return ca.currentAvailableFunds(ctx, channelID, types.NewInt(0))
}
// Merge all pending requests into one.
@@ -199,7 +199,7 @@ func (ca *channelAccessor) processQueue(channelID string) (*api.ChannelAvailable
if amt.IsZero() {
// Note: The amount can be zero if requests are cancelled as we're
// building the mergedFundsReq
return ca.currentAvailableFunds(channelID, amt)
return ca.currentAvailableFunds(ctx, channelID, amt)
}
res := ca.processTask(merged.ctx, amt)
@@ -209,7 +209,7 @@ func (ca *channelAccessor) processQueue(channelID string) (*api.ChannelAvailable
if res == nil {
// Stop processing the fundsReqQueue and wait. When the event occurs it will
// call processQueue() again
return ca.currentAvailableFunds(channelID, amt)
return ca.currentAvailableFunds(ctx, channelID, amt)
}
// Finished processing so clear the queue
@@ -218,7 +218,7 @@ func (ca *channelAccessor) processQueue(channelID string) (*api.ChannelAvailable
// Call the task callback with its results
merged.onComplete(res)
return ca.currentAvailableFunds(channelID, types.NewInt(0))
return ca.currentAvailableFunds(ctx, channelID, types.NewInt(0))
}
// filterQueue filters cancelled requests out of the queue
@@ -255,12 +255,12 @@ func (ca *channelAccessor) queueSize() int {
// msgWaitComplete is called when the message for a previous task is confirmed
// or there is an error.
func (ca *channelAccessor) msgWaitComplete(mcid cid.Cid, err error) {
func (ca *channelAccessor) msgWaitComplete(ctx context.Context, mcid cid.Cid, err error) {
ca.lk.Lock()
defer ca.lk.Unlock()
// Save the message result to the store
dserr := ca.store.SaveMessageResult(mcid, err)
dserr := ca.store.SaveMessageResult(ctx, mcid, err)
if dserr != nil {
log.Errorf("saving message result: %s", dserr)
}
@@ -271,16 +271,16 @@ func (ca *channelAccessor) msgWaitComplete(mcid cid.Cid, err error) {
// The queue may have been waiting for msg completion to proceed, so
// process the next queue item
if len(ca.fundsReqQueue) > 0 {
go ca.processQueue("") // nolint: errcheck
go ca.processQueue(ctx, "") // nolint: errcheck
}
}
func (ca *channelAccessor) currentAvailableFunds(channelID string, queuedAmt types.BigInt) (*api.ChannelAvailableFunds, error) {
func (ca *channelAccessor) currentAvailableFunds(ctx context.Context, channelID string, queuedAmt types.BigInt) (*api.ChannelAvailableFunds, error) {
if len(channelID) == 0 {
return nil, nil
}
channelInfo, err := ca.store.ByChannelID(channelID)
channelInfo, err := ca.store.ByChannelID(ctx, channelID)
if err != nil {
return nil, err
}
@@ -302,7 +302,7 @@ func (ca *channelAccessor) currentAvailableFunds(channelID string, queuedAmt typ
return nil, err
}
laneStates, err := ca.laneState(pchState, ch)
laneStates, err := ca.laneState(ctx, pchState, ch)
if err != nil {
return nil, err
}
@@ -337,7 +337,7 @@ func (ca *channelAccessor) processTask(ctx context.Context, amt types.BigInt) *p
// Get the payment channel for the from/to addresses.
// Note: It's ok if we get ErrChannelNotTracked. It just means we need to
// create a channel.
channelInfo, err := ca.store.OutboundActiveByFromTo(ca.from, ca.to)
channelInfo, err := ca.store.OutboundActiveByFromTo(ctx, ca.from, ca.to)
if err != nil && err != ErrChannelNotTracked {
return &paychFundsRes{err: err}
}
@@ -393,26 +393,26 @@ func (ca *channelAccessor) createPaych(ctx context.Context, amt types.BigInt) (c
mcid := smsg.Cid()
// Create a new channel in the store
ci, err := ca.store.CreateChannel(ca.from, ca.to, mcid, amt)
ci, err := ca.store.CreateChannel(ctx, ca.from, ca.to, mcid, amt)
if err != nil {
log.Errorf("creating channel: %s", err)
return cid.Undef, err
}
// Wait for the channel to be created on chain
go ca.waitForPaychCreateMsg(ci.ChannelID, mcid)
go ca.waitForPaychCreateMsg(ctx, ci.ChannelID, mcid)
return mcid, nil
}
// waitForPaychCreateMsg waits for mcid to appear on chain and stores the robust address of the
// created payment channel
func (ca *channelAccessor) waitForPaychCreateMsg(channelID string, mcid cid.Cid) {
err := ca.waitPaychCreateMsg(channelID, mcid)
ca.msgWaitComplete(mcid, err)
func (ca *channelAccessor) waitForPaychCreateMsg(ctx context.Context, channelID string, mcid cid.Cid) {
err := ca.waitPaychCreateMsg(ctx, channelID, mcid)
ca.msgWaitComplete(ctx, mcid, err)
}
func (ca *channelAccessor) waitPaychCreateMsg(channelID string, mcid cid.Cid) error {
func (ca *channelAccessor) waitPaychCreateMsg(ctx context.Context, channelID string, mcid cid.Cid) error {
mwait, err := ca.api.StateWaitMsg(ca.chctx, mcid, build.MessageConfidence, api.LookbackNoLimit, true)
if err != nil {
log.Errorf("wait msg: %v", err)
@@ -425,7 +425,7 @@ func (ca *channelAccessor) waitPaychCreateMsg(channelID string, mcid cid.Cid) er
defer ca.lk.Unlock()
// Channel creation failed, so remove the channel from the datastore
dserr := ca.store.RemoveChannel(channelID)
dserr := ca.store.RemoveChannel(ctx, channelID)
if dserr != nil {
log.Errorf("failed to remove channel %s: %s", channelID, dserr)
}
@@ -449,7 +449,7 @@ func (ca *channelAccessor) waitPaychCreateMsg(channelID string, mcid cid.Cid) er
defer ca.lk.Unlock()
// Store robust address of channel
ca.mutateChannelInfo(channelID, func(channelInfo *ChannelInfo) {
ca.mutateChannelInfo(ctx, channelID, func(channelInfo *ChannelInfo) {
channelInfo.Channel = &decodedReturn.RobustAddress
channelInfo.Amount = channelInfo.PendingAmount
channelInfo.PendingAmount = big.NewInt(0)
@@ -475,30 +475,30 @@ func (ca *channelAccessor) addFunds(ctx context.Context, channelInfo *ChannelInf
mcid := smsg.Cid()
// Store the add funds message CID on the channel
ca.mutateChannelInfo(channelInfo.ChannelID, func(ci *ChannelInfo) {
ca.mutateChannelInfo(ctx, channelInfo.ChannelID, func(ci *ChannelInfo) {
ci.PendingAmount = amt
ci.AddFundsMsg = &mcid
})
// Store a reference from the message CID to the channel, so that we can
// look up the channel from the message CID
err = ca.store.SaveNewMessage(channelInfo.ChannelID, mcid)
err = ca.store.SaveNewMessage(ctx, channelInfo.ChannelID, mcid)
if err != nil {
log.Errorf("saving add funds message CID %s: %s", mcid, err)
}
go ca.waitForAddFundsMsg(channelInfo.ChannelID, mcid)
go ca.waitForAddFundsMsg(ctx, channelInfo.ChannelID, mcid)
return &mcid, nil
}
// waitForAddFundsMsg waits for mcid to appear on chain and returns error, if any
func (ca *channelAccessor) waitForAddFundsMsg(channelID string, mcid cid.Cid) {
err := ca.waitAddFundsMsg(channelID, mcid)
ca.msgWaitComplete(mcid, err)
func (ca *channelAccessor) waitForAddFundsMsg(ctx context.Context, channelID string, mcid cid.Cid) {
err := ca.waitAddFundsMsg(ctx, channelID, mcid)
ca.msgWaitComplete(ctx, mcid, err)
}
func (ca *channelAccessor) waitAddFundsMsg(channelID string, mcid cid.Cid) error {
func (ca *channelAccessor) waitAddFundsMsg(ctx context.Context, channelID string, mcid cid.Cid) error {
mwait, err := ca.api.StateWaitMsg(ca.chctx, mcid, build.MessageConfidence, api.LookbackNoLimit, true)
if err != nil {
log.Error(err)
@@ -512,7 +512,7 @@ func (ca *channelAccessor) waitAddFundsMsg(channelID string, mcid cid.Cid) error
ca.lk.Lock()
defer ca.lk.Unlock()
ca.mutateChannelInfo(channelID, func(channelInfo *ChannelInfo) {
ca.mutateChannelInfo(ctx, channelID, func(channelInfo *ChannelInfo) {
channelInfo.PendingAmount = big.NewInt(0)
channelInfo.AddFundsMsg = nil
})
@@ -524,7 +524,7 @@ func (ca *channelAccessor) waitAddFundsMsg(channelID string, mcid cid.Cid) error
defer ca.lk.Unlock()
// Store updated amount
ca.mutateChannelInfo(channelID, func(channelInfo *ChannelInfo) {
ca.mutateChannelInfo(ctx, channelID, func(channelInfo *ChannelInfo) {
channelInfo.Amount = types.BigAdd(channelInfo.Amount, channelInfo.PendingAmount)
channelInfo.PendingAmount = big.NewInt(0)
channelInfo.AddFundsMsg = nil
@@ -534,8 +534,8 @@ func (ca *channelAccessor) waitAddFundsMsg(channelID string, mcid cid.Cid) error
}
// Change the state of the channel in the store
func (ca *channelAccessor) mutateChannelInfo(channelID string, mutate func(*ChannelInfo)) {
channelInfo, err := ca.store.ByChannelID(channelID)
func (ca *channelAccessor) mutateChannelInfo(ctx context.Context, channelID string, mutate func(*ChannelInfo)) {
channelInfo, err := ca.store.ByChannelID(ctx, channelID)
// If there's an error reading or writing to the store just log an error.
// For now we're assuming it's unlikely to happen in practice.
@@ -549,7 +549,7 @@ func (ca *channelAccessor) mutateChannelInfo(channelID string, mutate func(*Chan
mutate(channelInfo)
err = ca.store.putChannelInfo(channelInfo)
err = ca.store.putChannelInfo(ctx, channelInfo)
if err != nil {
log.Errorf("Error writing channel info to store: %s", err)
}
@@ -560,8 +560,8 @@ func (ca *channelAccessor) mutateChannelInfo(channelID string, mutate func(*Chan
// messages.
// Outstanding messages can occur if a create / add funds message was sent and
// then the system was shut down or crashed before the result was received.
func (pm *Manager) restartPending() error {
cis, err := pm.store.WithPendingAddFunds()
func (pm *Manager) restartPending(ctx context.Context) error {
cis, err := pm.store.WithPendingAddFunds(ctx)
if err != nil {
return err
}
@@ -575,16 +575,16 @@ func (pm *Manager) restartPending() error {
if err != nil {
return xerrors.Errorf("error initializing payment channel manager %s -> %s: %s", ci.Control, ci.Target, err)
}
go ca.waitForPaychCreateMsg(ci.ChannelID, *ci.CreateMsg)
go ca.waitForPaychCreateMsg(ctx, ci.ChannelID, *ci.CreateMsg)
return nil
})
} else if ci.AddFundsMsg != nil {
group.Go(func() error {
ca, err := pm.accessorByAddress(*ci.Channel)
ca, err := pm.accessorByAddress(ctx, *ci.Channel)
if err != nil {
return xerrors.Errorf("error initializing payment channel manager %s: %s", ci.Channel, err)
}
go ca.waitForAddFundsMsg(ci.ChannelID, *ci.AddFundsMsg)
go ca.waitForAddFundsMsg(ctx, ci.ChannelID, *ci.AddFundsMsg)
return nil
})
}
@@ -598,7 +598,7 @@ func (ca *channelAccessor) getPaychWaitReady(ctx context.Context, mcid cid.Cid)
ca.lk.Lock()
// First check if the message has completed
msgInfo, err := ca.store.GetMessage(mcid)
msgInfo, err := ca.store.GetMessage(ctx, mcid)
if err != nil {
ca.lk.Unlock()
@@ -617,7 +617,7 @@ func (ca *channelAccessor) getPaychWaitReady(ctx context.Context, mcid cid.Cid)
ca.lk.Unlock()
// Get the channel address
ci, err := ca.store.ByMessageCid(mcid)
ci, err := ca.store.ByMessageCid(ctx, mcid)
if err != nil {
return address.Undef, err
}
@@ -660,7 +660,7 @@ func (ca *channelAccessor) msgPromise(ctx context.Context, mcid cid.Cid) chan on
res := onMsgRes{err: err}
if res.err == nil {
// Get the channel associated with the message cid
ci, err := ca.store.ByMessageCid(mcid)
ci, err := ca.store.ByMessageCid(ctx, mcid)
if err != nil {
res.err = err
} else {
@@ -689,6 +689,6 @@ func (ca *channelAccessor) msgPromise(ctx context.Context, mcid cid.Cid) chan on
return promise
}
func (ca *channelAccessor) availableFunds(channelID string) (*api.ChannelAvailableFunds, error) {
return ca.processQueue(channelID)
func (ca *channelAccessor) availableFunds(ctx context.Context, channelID string) (*api.ChannelAvailableFunds, error) {
return ca.processQueue(ctx, channelID)
}
paychmgr/store.go
+43 -42
View File
@@ -2,6 +2,7 @@ package paychmgr
import (
"bytes"
"context"
"errors"
"fmt"
@@ -157,26 +158,26 @@ func (ci *ChannelInfo) wasVoucherSubmitted(sv *paych.SignedVoucher) (bool, error
// TrackChannel stores a channel, returning an error if the channel was already
// being tracked
func (ps *Store) TrackChannel(ci *ChannelInfo) (*ChannelInfo, error) {
_, err := ps.ByAddress(*ci.Channel)
func (ps *Store) TrackChannel(ctx context.Context, ci *ChannelInfo) (*ChannelInfo, error) {
_, err := ps.ByAddress(ctx, *ci.Channel)
switch err {
default:
return nil, err
case nil:
return nil, fmt.Errorf("already tracking channel: %s", ci.Channel)
case ErrChannelNotTracked:
err = ps.putChannelInfo(ci)
err = ps.putChannelInfo(ctx, ci)
if err != nil {
return nil, err
}
return ps.ByAddress(*ci.Channel)
return ps.ByAddress(ctx, *ci.Channel)
}
}
// ListChannels returns the addresses of all channels that have been created
func (ps *Store) ListChannels() ([]address.Address, error) {
cis, err := ps.findChans(func(ci *ChannelInfo) bool {
func (ps *Store) ListChannels(ctx context.Context) ([]address.Address, error) {
cis, err := ps.findChans(ctx, func(ci *ChannelInfo) bool {
return ci.Channel != nil
}, 0)
if err != nil {
@@ -193,8 +194,8 @@ func (ps *Store) ListChannels() ([]address.Address, error) {
// findChan finds a single channel using the given filter.
// If there isn't a channel that matches the filter, returns ErrChannelNotTracked
func (ps *Store) findChan(filter func(ci *ChannelInfo) bool) (*ChannelInfo, error) {
cis, err := ps.findChans(filter, 1)
func (ps *Store) findChan(ctx context.Context, filter func(ci *ChannelInfo) bool) (*ChannelInfo, error) {
cis, err := ps.findChans(ctx, filter, 1)
if err != nil {
return nil, err
}
@@ -208,8 +209,8 @@ func (ps *Store) findChan(filter func(ci *ChannelInfo) bool) (*ChannelInfo, erro
// findChans loops over all channels, only including those that pass the filter.
// max is the maximum number of channels to return. Set to zero to return unlimited channels.
func (ps *Store) findChans(filter func(*ChannelInfo) bool, max int) ([]ChannelInfo, error) {
res, err := ps.ds.Query(dsq.Query{Prefix: dsKeyChannelInfo})
func (ps *Store) findChans(ctx context.Context, filter func(*ChannelInfo) bool, max int) ([]ChannelInfo, error) {
res, err := ps.ds.Query(ctx, dsq.Query{Prefix: dsKeyChannelInfo})
if err != nil {
return nil, err
}
@@ -251,8 +252,8 @@ func (ps *Store) findChans(filter func(*ChannelInfo) bool, max int) ([]ChannelIn
}
// AllocateLane allocates a new lane for the given channel
func (ps *Store) AllocateLane(ch address.Address) (uint64, error) {
ci, err := ps.ByAddress(ch)
func (ps *Store) AllocateLane(ctx context.Context, ch address.Address) (uint64, error) {
ci, err := ps.ByAddress(ctx, ch)
if err != nil {
return 0, err
}
@@ -260,12 +261,12 @@ func (ps *Store) AllocateLane(ch address.Address) (uint64, error) {
out := ci.NextLane
ci.NextLane++
return out, ps.putChannelInfo(ci)
return out, ps.putChannelInfo(ctx, ci)
}
// VouchersForPaych gets the vouchers for the given channel
func (ps *Store) VouchersForPaych(ch address.Address) ([]*VoucherInfo, error) {
ci, err := ps.ByAddress(ch)
func (ps *Store) VouchersForPaych(ctx context.Context, ch address.Address) ([]*VoucherInfo, error) {
ci, err := ps.ByAddress(ctx, ch)
if err != nil {
return nil, err
}
@@ -273,17 +274,17 @@ func (ps *Store) VouchersForPaych(ch address.Address) ([]*VoucherInfo, error) {
return ci.Vouchers, nil
}
func (ps *Store) MarkVoucherSubmitted(ci *ChannelInfo, sv *paych.SignedVoucher) error {
func (ps *Store) MarkVoucherSubmitted(ctx context.Context, ci *ChannelInfo, sv *paych.SignedVoucher) error {
err := ci.markVoucherSubmitted(sv)
if err != nil {
return err
}
return ps.putChannelInfo(ci)
return ps.putChannelInfo(ctx, ci)
}
// ByAddress gets the channel that matches the given address
func (ps *Store) ByAddress(addr address.Address) (*ChannelInfo, error) {
return ps.findChan(func(ci *ChannelInfo) bool {
func (ps *Store) ByAddress(ctx context.Context, addr address.Address) (*ChannelInfo, error) {
return ps.findChan(ctx, func(ci *ChannelInfo) bool {
return ci.Channel != nil && *ci.Channel == addr
})
}
@@ -307,7 +308,7 @@ func dskeyForMsg(mcid cid.Cid) datastore.Key {
}
// SaveNewMessage is called when a message is sent
func (ps *Store) SaveNewMessage(channelID string, mcid cid.Cid) error {
func (ps *Store) SaveNewMessage(ctx context.Context, channelID string, mcid cid.Cid) error {
k := dskeyForMsg(mcid)
b, err := cborrpc.Dump(&MsgInfo{ChannelID: channelID, MsgCid: mcid})
@@ -315,12 +316,12 @@ func (ps *Store) SaveNewMessage(channelID string, mcid cid.Cid) error {
return err
}
return ps.ds.Put(k, b)
return ps.ds.Put(ctx, k, b)
}
// SaveMessageResult is called when the result of a message is received
func (ps *Store) SaveMessageResult(mcid cid.Cid, msgErr error) error {
minfo, err := ps.GetMessage(mcid)
func (ps *Store) SaveMessageResult(ctx context.Context, mcid cid.Cid, msgErr error) error {
minfo, err := ps.GetMessage(ctx, mcid)
if err != nil {
return err
}
@@ -336,17 +337,17 @@ func (ps *Store) SaveMessageResult(mcid cid.Cid, msgErr error) error {
return err
}
return ps.ds.Put(k, b)
return ps.ds.Put(ctx, k, b)
}
// ByMessageCid gets the channel associated with a message
func (ps *Store) ByMessageCid(mcid cid.Cid) (*ChannelInfo, error) {
minfo, err := ps.GetMessage(mcid)
func (ps *Store) ByMessageCid(ctx context.Context, mcid cid.Cid) (*ChannelInfo, error) {
minfo, err := ps.GetMessage(ctx, mcid)
if err != nil {
return nil, err
}
ci, err := ps.findChan(func(ci *ChannelInfo) bool {
ci, err := ps.findChan(ctx, func(ci *ChannelInfo) bool {
return ci.ChannelID == minfo.ChannelID
})
if err != nil {
@@ -357,10 +358,10 @@ func (ps *Store) ByMessageCid(mcid cid.Cid) (*ChannelInfo, error) {
}
// GetMessage gets the message info for a given message CID
func (ps *Store) GetMessage(mcid cid.Cid) (*MsgInfo, error) {
func (ps *Store) GetMessage(ctx context.Context, mcid cid.Cid) (*MsgInfo, error) {
k := dskeyForMsg(mcid)
val, err := ps.ds.Get(k)
val, err := ps.ds.Get(ctx, k)
if err != nil {
return nil, err
}
@@ -375,8 +376,8 @@ func (ps *Store) GetMessage(mcid cid.Cid) (*MsgInfo, error) {
// OutboundActiveByFromTo looks for outbound channels that have not been
// settled, with the given from / to addresses
func (ps *Store) OutboundActiveByFromTo(from address.Address, to address.Address) (*ChannelInfo, error) {
return ps.findChan(func(ci *ChannelInfo) bool {
func (ps *Store) OutboundActiveByFromTo(ctx context.Context, from address.Address, to address.Address) (*ChannelInfo, error) {
return ps.findChan(ctx, func(ci *ChannelInfo) bool {
if ci.Direction != DirOutbound {
return false
}
@@ -390,8 +391,8 @@ func (ps *Store) OutboundActiveByFromTo(from address.Address, to address.Address
// WithPendingAddFunds is used on startup to find channels for which a
// create channel or add funds message has been sent, but lotus shut down
// before the response was received.
func (ps *Store) WithPendingAddFunds() ([]ChannelInfo, error) {
return ps.findChans(func(ci *ChannelInfo) bool {
func (ps *Store) WithPendingAddFunds(ctx context.Context) ([]ChannelInfo, error) {
return ps.findChans(ctx, func(ci *ChannelInfo) bool {
if ci.Direction != DirOutbound {
return false
}
@@ -400,10 +401,10 @@ func (ps *Store) WithPendingAddFunds() ([]ChannelInfo, error) {
}
// ByChannelID gets channel info by channel ID
func (ps *Store) ByChannelID(channelID string) (*ChannelInfo, error) {
func (ps *Store) ByChannelID(ctx context.Context, channelID string) (*ChannelInfo, error) {
var stored ChannelInfo
res, err := ps.ds.Get(dskeyForChannel(channelID))
res, err := ps.ds.Get(ctx, dskeyForChannel(channelID))
if err != nil {
if err == datastore.ErrNotFound {
return nil, ErrChannelNotTracked
@@ -415,7 +416,7 @@ func (ps *Store) ByChannelID(channelID string) (*ChannelInfo, error) {
}
// CreateChannel creates an outbound channel for the given from / to
func (ps *Store) CreateChannel(from address.Address, to address.Address, createMsgCid cid.Cid, amt types.BigInt) (*ChannelInfo, error) {
func (ps *Store) CreateChannel(ctx context.Context, from address.Address, to address.Address, createMsgCid cid.Cid, amt types.BigInt) (*ChannelInfo, error) {
ci := &ChannelInfo{
Direction: DirOutbound,
NextLane: 0,
@@ -426,13 +427,13 @@ func (ps *Store) CreateChannel(from address.Address, to address.Address, createM
}
// Save the new channel
err := ps.putChannelInfo(ci)
err := ps.putChannelInfo(ctx, ci)
if err != nil {
return nil, err
}
// Save a reference to the create message
err = ps.SaveNewMessage(ci.ChannelID, createMsgCid)
err = ps.SaveNewMessage(ctx, ci.ChannelID, createMsgCid)
if err != nil {
return nil, err
}
@@ -441,8 +442,8 @@ func (ps *Store) CreateChannel(from address.Address, to address.Address, createM
}
// RemoveChannel removes the channel with the given channel ID
func (ps *Store) RemoveChannel(channelID string) error {
return ps.ds.Delete(dskeyForChannel(channelID))
func (ps *Store) RemoveChannel(ctx context.Context, channelID string) error {
return ps.ds.Delete(ctx, dskeyForChannel(channelID))
}
// The datastore key used to identify the channel info
@@ -451,7 +452,7 @@ func dskeyForChannel(channelID string) datastore.Key {
}
// putChannelInfo stores the channel info in the datastore
func (ps *Store) putChannelInfo(ci *ChannelInfo) error {
func (ps *Store) putChannelInfo(ctx context.Context, ci *ChannelInfo) error {
if len(ci.ChannelID) == 0 {
ci.ChannelID = uuid.New().String()
}
@@ -462,7 +463,7 @@ func (ps *Store) putChannelInfo(ci *ChannelInfo) error {
return err
}
return ps.ds.Put(k, b)
return ps.ds.Put(ctx, k, b)
}
// TODO: This is a hack to get around not being able to CBOR marshall a nil