contractcourt: update startup log reads to take optional read tx

When we startup large nodes, each arbitrator needs a minimum of 2 db
transactions to lookup its current state. This commit updates our arb
log interface and implementation to allow passing in an optional read
transaction. This will allow us to batch these reads in a single db tx
in the following commit.

Author: carlaKC

contractcourt/briefcase.go

@@ -54,8 +54,10 @@ type ArbitratorLog interface {
 	// TODO(roasbeef): document on interface the errors expected to be
 	// returned
 
-	// CurrentState returns the current state of the ChannelArbitrator.
-	CurrentState() (ArbitratorState, error)
+	// CurrentState returns the current state of the ChannelArbitrator. It
+	// takes an optional database transaction, which will be used if it is
+	// non-nil, otherwise the lookup will be done in its own transaction.
+	CurrentState(tx kvdb.RTx) (ArbitratorState, error)
 
 	// CommitState persists, the current state of the chain attendant.
 	CommitState(ArbitratorState) error
@@ -96,8 +98,10 @@ type ArbitratorLog interface {
 	InsertConfirmedCommitSet(c *CommitSet) error
 
 	// FetchConfirmedCommitSet fetches the known confirmed active HTLC set
-	// from the database.
-	FetchConfirmedCommitSet() (*CommitSet, error)
+	// from the database. It takes an optional database transaction, which
+	// will be used if it is non-nil, otherwise the lookup will be done in
+	// its own transaction.
+	FetchConfirmedCommitSet(tx kvdb.RTx) (*CommitSet, error)
 
 	// FetchChainActions attempts to fetch the set of previously stored
 	// chain actions. We'll use this upon restart to properly advance our
@@ -412,32 +416,47 @@ func (b *boltArbitratorLog) writeResolver(contractBucket kvdb.RwBucket,
 	return contractBucket.Put(resKey, buf.Bytes())
 }
 
-// CurrentState returns the current state of the ChannelArbitrator.
+// CurrentState returns the current state of the ChannelArbitrator. It takes an
+// optional database transaction, which will be used if it is non-nil, otherwise
+// the lookup will be done in its own transaction.
 //
 // NOTE: Part of the ContractResolver interface.
-func (b *boltArbitratorLog) CurrentState() (ArbitratorState, error) {
-	var s ArbitratorState
-	err := kvdb.View(b.db, func(tx kvdb.RTx) error {
-		scopeBucket := tx.ReadBucket(b.scopeKey[:])
-		if scopeBucket == nil {
-			return errScopeBucketNoExist
-		}
+func (b *boltArbitratorLog) CurrentState(tx kvdb.RTx) (ArbitratorState, error) {
+	var (
+		s   ArbitratorState
+		err error
+	)
 
-		stateBytes := scopeBucket.Get(stateKey)
-		if stateBytes == nil {
-			return nil
-		}
+	if tx != nil {
+		s, err = b.currentState(tx)
+	} else {
+		err = kvdb.View(b.db, func(tx kvdb.RTx) error {
+			s, err = b.currentState(tx)
+			return err
+		})
+	}
 
-		s = ArbitratorState(stateBytes[0])
-		return nil
-	})
 	if err != nil && err != errScopeBucketNoExist {
 		return s, err
 	}
 
 	return s, nil
 }
 
+func (b *boltArbitratorLog) currentState(tx kvdb.RTx) (ArbitratorState, error) {
+	scopeBucket := tx.ReadBucket(b.scopeKey[:])
+	if scopeBucket == nil {
+		return 0, errScopeBucketNoExist
+	}
+
+	stateBytes := scopeBucket.Get(stateKey)
+	if stateBytes == nil {
+		return 0, nil
+	}
+
+	return ArbitratorState(stateBytes[0]), nil
+}
+
 // CommitState persists, the current state of the chain attendant.
 //
 // NOTE: Part of the ContractResolver interface.
@@ -843,29 +862,20 @@ func (b *boltArbitratorLog) InsertConfirmedCommitSet(c *CommitSet) error {
 }
 
 // FetchConfirmedCommitSet fetches the known confirmed active HTLC set from the
-// database.
+// database. It takes an optional database transaction, which will be used if it
+// is non-nil, otherwise the lookup will be done in its own transaction.
 //
 // NOTE: Part of the ContractResolver interface.
-func (b *boltArbitratorLog) FetchConfirmedCommitSet() (*CommitSet, error) {
+func (b *boltArbitratorLog) FetchConfirmedCommitSet(tx kvdb.RTx) (*CommitSet, error) {
+	if tx != nil {
+		return b.fetchConfirmedCommitSet(tx)
+	}
+
 	var c *CommitSet
 	err := kvdb.View(b.db, func(tx kvdb.RTx) error {
-		scopeBucket := tx.ReadBucket(b.scopeKey[:])
-		if scopeBucket == nil {
-			return errScopeBucketNoExist
-		}
-
-		commitSetBytes := scopeBucket.Get(commitSetKey)
-		if commitSetBytes == nil {
-			return errNoCommitSet
-		}
-
-		commitSet, err := decodeCommitSet(bytes.NewReader(commitSetBytes))
-		if err != nil {
-			return err
-		}
-
-		c = commitSet
-		return nil
+		var err error
+		c, err = b.fetchConfirmedCommitSet(tx)
+		return err
 	})
 	if err != nil {
 		return nil, err
@@ -874,6 +884,22 @@ func (b *boltArbitratorLog) FetchConfirmedCommitSet() (*CommitSet, error) {
 	return c, nil
 }
 
+func (b *boltArbitratorLog) fetchConfirmedCommitSet(tx kvdb.RTx) (*CommitSet,
+	error) {
+
+	scopeBucket := tx.ReadBucket(b.scopeKey[:])
+	if scopeBucket == nil {
+		return nil, errScopeBucketNoExist
+	}
+
+	commitSetBytes := scopeBucket.Get(commitSetKey)
+	if commitSetBytes == nil {
+		return nil, errNoCommitSet
+	}
+
+	return decodeCommitSet(bytes.NewReader(commitSetBytes))
+}
+
 // WipeHistory is to be called ONLY once *all* contracts have been fully
 // resolved, and the channel closure if finalized. This method will delete all
 // on-disk state within the persistent log.

contractcourt/briefcase_test.go

@@ -611,7 +611,7 @@ func TestStateMutation(t *testing.T) {
 	defer cleanUp()
 
 	// The default state of an arbitrator should be StateDefault.
-	arbState, err := testLog.CurrentState()
+	arbState, err := testLog.CurrentState(nil)
 	if err != nil {
 		t.Fatalf("unable to read arb state: %v", err)
 	}
@@ -625,7 +625,7 @@ func TestStateMutation(t *testing.T) {
 	if err := testLog.CommitState(StateFullyResolved); err != nil {
 		t.Fatalf("unable to write state: %v", err)
 	}
-	arbState, err = testLog.CurrentState()
+	arbState, err = testLog.CurrentState(nil)
 	if err != nil {
 		t.Fatalf("unable to read arb state: %v", err)
 	}
@@ -643,7 +643,7 @@ func TestStateMutation(t *testing.T) {
 
 	// If we try to query for the state again, we should get the default
 	// state again.
-	arbState, err = testLog.CurrentState()
+	arbState, err = testLog.CurrentState(nil)
 	if err != nil {
 		t.Fatalf("unable to query current state: %v", err)
 	}
@@ -687,11 +687,11 @@ func TestScopeIsolation(t *testing.T) {
 
 	// Querying each log, the states should be the prior one we set, and be
 	// disjoint.
-	log1State, err := testLog1.CurrentState()
+	log1State, err := testLog1.CurrentState(nil)
 	if err != nil {
 		t.Fatalf("unable to read arb state: %v", err)
 	}
-	log2State, err := testLog2.CurrentState()
+	log2State, err := testLog2.CurrentState(nil)
 	if err != nil {
 		t.Fatalf("unable to read arb state: %v", err)
 	}
@@ -752,7 +752,7 @@ func TestCommitSetStorage(t *testing.T) {
 				t.Fatalf("unable to write commit set: %v", err)
 			}
 
-			diskCommitSet, err := testLog.FetchConfirmedCommitSet()
+			diskCommitSet, err := testLog.FetchConfirmedCommitSet(nil)
 			if err != nil {
 				t.Fatalf("unable to read commit set: %v", err)
 			}

contractcourt/chain_arbitrator.go

@@ -616,7 +616,7 @@ func (c *ChainArbitrator) Start() error {
 	// Launch all the goroutines for each arbitrator so they can carry out
 	// their duties.
 	for _, arbitrator := range c.activeChannels {
-		if err := arbitrator.Start(); err != nil {
+		if err := arbitrator.Start(nil); err != nil {
 			c.Stop()
 			return err
 		}
@@ -1036,7 +1036,7 @@ func (c *ChainArbitrator) WatchNewChannel(newChan *channeldb.OpenChannel) error
 	// arbitrators, then launch it.
 	c.activeChannels[chanPoint] = channelArb
 
-	if err := channelArb.Start(); err != nil {
+	if err := channelArb.Start(nil); err != nil {
 		return err
 	}
 

contractcourt/channel_arbitrator.go

@@ -376,7 +376,7 @@ func NewChannelArbitrator(cfg ChannelArbitratorConfig,
 }
 
 // Start starts all the goroutines that the ChannelArbitrator needs to operate.
-func (c *ChannelArbitrator) Start() error {
+func (c *ChannelArbitrator) Start(tx kvdb.RTx) error {
 	if !atomic.CompareAndSwapInt32(&c.started, 0, 1) {
 		return nil
 	}
@@ -394,7 +394,7 @@ func (c *ChannelArbitrator) Start() error {
 
 	// First, we'll read our last state from disk, so our internal state
 	// machine can act accordingly.
-	c.state, err = c.log.CurrentState()
+	c.state, err = c.log.CurrentState(tx)
 	if err != nil {
 		return err
 	}
@@ -450,7 +450,7 @@ func (c *ChannelArbitrator) Start() error {
 	// older nodes, this won't be found at all, and will rely on the
 	// existing written chain actions. Additionally, if this channel hasn't
 	// logged any actions in the log, then this field won't be present.
-	commitSet, err := c.log.FetchConfirmedCommitSet()
+	commitSet, err := c.log.FetchConfirmedCommitSet(tx)
 	if err != nil && err != errNoCommitSet && err != errScopeBucketNoExist {
 		return err
 	}