diff --git a/network/node/register.go b/network/node/register.go index 1576fe89a027cbba1a29ef374a84c809848e0d57..274703a8d1fb68f001e0ab692a8c726243ed38dc 100644 --- a/network/node/register.go +++ b/network/node/register.go @@ -39,6 +39,10 @@ import ( "time" ) + +const maxAttempts = 5 +var delayTable = [5]time.Duration{0,5*time.Second,30*time.Second,60*time.Second,120*time.Second} + type RegisterNodeCommsInterface interface { SendRequestClientKeyMessage(host *connect.Host, message *pb.SignedClientKeyRequest) (*pb.SignedKeyResponse, error) @@ -49,12 +53,16 @@ func StartRegistration(sender *gateway.Sender, session *storage.Session, rngGen multi := stoppable.NewMulti("NodeRegistrations") + inProgess := &sync.Map{} + // we are relying on the in progress check to + // ensure there is only a single operator at a time, as a result this is a map of ID -> int + attempts := &sync.Map{} for i := uint(0); i < numParallel; i++ { stop := stoppable.NewSingle(fmt.Sprintf("NodeRegistration %d", i)) - go registerNodes(sender, session, rngGen, comms, stop, c, inProgess) + go registerNodes(sender, session, rngGen, comms, stop, c, inProgess, attempts) multi.Add(stop) } @@ -63,7 +71,7 @@ func StartRegistration(sender *gateway.Sender, session *storage.Session, rngGen func registerNodes(sender *gateway.Sender, session *storage.Session, rngGen *fastRNG.StreamGenerator, comms RegisterNodeCommsInterface, - stop *stoppable.Single, c chan network.NodeGateway, inProgress *sync.Map) { + stop *stoppable.Single, c chan network.NodeGateway, inProgress, attempts *sync.Map) { u := session.User() regSignature := u.GetTransmissionRegistrationValidationSignature() // Timestamp in which user has registered with registration @@ -85,6 +93,14 @@ func registerNodes(sender *gateway.Sender, session *storage.Session, if _, operating := inProgress.LoadOrStore(nidStr, struct{}{}); operating { continue } + + //keep track of how many times this has been attempted + numAttempts := uint(1) + if nunAttemptsInterface, hasValue := attempts.LoadOrStore(nidStr, numAttempts); hasValue{ + numAttempts = nunAttemptsInterface.(uint) + attempts.Store(nidStr, numAttempts+1) + } + // No need to register with stale nodes if isStale := gw.Node.Status == ndf.Stale; isStale { jww.DEBUG.Printf("Skipping registration with stale node %s", nidStr) @@ -95,12 +111,22 @@ func registerNodes(sender *gateway.Sender, session *storage.Session, inProgress.Delete(nidStr) if err != nil { jww.ERROR.Printf("Failed to register node: %+v", err) + //if we have not reached the attempt limit for this gateway, send it back into the channel to retry + if numAttempts <maxAttempts{ + go func(){ + //delay the send for a backoff + time.Sleep(delayTable[numAttempts-1]) + c<-gw + }() + } } case <-t.C: } } } + + //registerWithNode serves as a helper for RegisterWithNodes // It registers a user with a specific in the client's ndf. func registerWithNode(sender *gateway.Sender, comms RegisterNodeCommsInterface,