connect: enforce contracts on forward create

Author: xcolwell

connect/connect_test.go

@@ -534,6 +534,8 @@ func testConnect(
 		switch contractTest {
 		case contractTestSymmetric, contractTestAsymmetric:
 			settings.ForwardEnforceActiveContracts = true
+		default:
+			settings.ForwardEnforceActiveContracts = false
 		}
 
 		exchange := NewExchange(ctx, host, service, block, hostToServicePorts, routes, settings)

connect/resident.go

@@ -144,7 +144,7 @@ func DefaultExchangeSettings() *ExchangeSettings {
 		ExchangeResidentWaitTimeout: exchangeResidentWaitTimeout,
 		ExchangeResidentPollTimeout: 15 * time.Second,
 
-		ForwardEnforceActiveContracts: false,
+		ForwardEnforceActiveContracts: true,
 
 		ExchangeChaosSettings: *DefaultExchangeChaosSettings(),
 		// default drain 300/minute
@@ -1583,37 +1583,36 @@ func (self *Resident) handleClientForward(path connect.TransferPath, transferFra
 
 	// FIXME deep packet inspection to look at the contract frames and verify contracts before forwarding
 
-	if self.exchange.settings.ForwardEnforceActiveContracts {
-		if !isAck(transferFrameBytes) {
-			hasActiveContract := self.residentContractManager.HasActiveContract(sourceId, destinationId)
-			if !hasActiveContract {
-				glog.Infof("[rf]abuse no active contract %s->%s\n", sourceId, destinationId)
-				// there is no active contract
-				// drop
-				self.abuseLimiter.delay()
-				return
-			}
-		}
-	}
-
 	c := func() bool {
 
 		nextForward := func() *ResidentForward {
+			if self.exchange.settings.ForwardEnforceActiveContracts {
+				// if !isAck(transferFrameBytes) {
+				hasActiveContract := self.residentContractManager.HasActiveContract(sourceId, destinationId)
+				if !hasActiveContract {
+					glog.Infof("[rf]abuse no active contract %s->%s\n", sourceId, destinationId)
+					// there is no active contract
+					// drop
+					self.abuseLimiter.delay()
+					return nil
+				}
+				// }
+			}
+
 			forward := NewResidentForward(self.ctx, self.exchange, destinationId)
 			go server.HandleError(func() {
-				forward.Run()
-
-				glog.V(1).Infof("[rf]close %s->%s\n", sourceId, destinationId)
-
-				// note we don't call close here because only the sender should call close
-				forward.Cancel()
-				func() {
+				defer func() {
 					self.stateLock.Lock()
 					defer self.stateLock.Unlock()
+					forward.Cancel()
 					if currentForward := self.forwards[destinationId]; forward == currentForward {
 						delete(self.forwards, destinationId)
 					}
 				}()
+				forward.Run()
+
+				glog.V(1).Infof("[rf]close %s->%s\n", sourceId, destinationId)
+				// note we don't call close here because only the sender should call close
 			}, forward.Cancel)
 			go server.HandleError(func() {
 				defer forward.Cancel()
@@ -1670,6 +1669,10 @@ func (self *Resident) handleClientForward(path connect.TransferPath, transferFra
 			forward = nextForward()
 		}
 
+		if forward == nil {
+			return false
+		}
+
 		select {
 		case <-self.ctx.Done():
 			return false

connect/resident_contract_manager.go

@@ -71,8 +71,17 @@ func (self *residentContractManager) HasActiveContract(sourceId server.Id, desti
 
 	next := func() (nextEntry *activeContractEntry) {
 		c := func() bool {
-			contractIds := model.GetOpenContractIdsWithNoPartialClose(self.ctx, sourceId, destinationId)
-			return 0 < len(contractIds)
+			contractIds1 := model.GetOpenContractIdsWithNoPartialClose(self.ctx, sourceId, destinationId)
+			if 0 < len(contractIds1) {
+				return true
+			}
+
+			contractIds2 := model.GetOpenContractIdsWithNoPartialClose(self.ctx, destinationId, sourceId)
+			if 0 < len(contractIds2) {
+				return true
+			}
+
+			return false
 		}
 		hasActiveContract := c()