Improve hub locking

.github/workflows/ci.yaml

@@ -115,10 +115,10 @@ jobs:
         working-directory: ./be1-go
 
     steps:
-      - name: Use go >= 1.19
+      - name: Use go >= 1.21
         uses: actions/setup-go@v3
         with:
-          go-version: ">=1.19"
+          go-version: ">=1.21"
 
       - name: Setup repo
         uses: actions/checkout@v3

.github/workflows/deploy.yaml

@@ -91,10 +91,10 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
-      - name: Use go >= 1.19
+      - name: Use go >= 1.21
         uses: actions/setup-go@v3
         with:
-          go-version: ">=1.19"
+          go-version: ">=1.21"
 
       - name: build
         run: |

.github/workflows/karate_be1-go.yaml

@@ -11,10 +11,10 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - name: Use go >= 1.19
+      - name: Use go >= 1.21
         uses: actions/setup-go@v3
         with:
-          go-version: ">=1.19"
+          go-version: ">=1.21"
 
       - name: Setup repo
         uses: actions/checkout@v3

be1-go/.gitignore

@@ -4,4 +4,6 @@ report.json
 coverage.out
 coverage.html
 configServer1.json
-configServer2.json
+configServer2.json
+configServer3.json
+configServer4.json

be1-go/channel/lao/mod.go

@@ -702,12 +702,12 @@ func (c *Channel) createAndSendLAOGreet() error {
 		return xerrors.Errorf("failed to marshal the organizer key: %v", err)
 	}
 
-	peers := []messagedata.Peer{}
+	peersInfo := c.hub.GetPeersInfo()
 
-	for _, info := range c.hub.GetPeersInfo() {
-		peers = append(peers, messagedata.Peer{
-			Address: info.ClientAddress,
-		})
+	peers := make([]messagedata.Peer, 0, len(peersInfo))
+
+	for _, info := range peersInfo {
+		peers = append(peers, messagedata.Peer{Address: info.ClientAddress})
 	}
 
 	msgData := messagedata.LaoGreet{

be1-go/configServer2.json

@@ -11,4 +11,4 @@
     "other-servers": [
         "localhost:9001"
     ]
-}
+}

be1-go/go.mod

@@ -1,6 +1,6 @@
 module popstellar
 
-go 1.19
+go 1.21
 
 require (
 	github.com/aaronarduino/goqrsvg v0.0.0-20220419053939-17e843f1dd40
@@ -20,7 +20,7 @@ require (
 	github.com/zitadel/oidc/v2 v2.1.2
 	go.dedis.ch/kyber/v3 v3.0.13
 	golang.org/x/exp v0.0.0-20230321023759-10a507213a29
-	golang.org/x/sync v0.1.0
+	golang.org/x/sync v0.4.0
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2
 	gopkg.in/yaml.v2 v2.2.3
 )
@@ -38,11 +38,11 @@ require (
 	github.com/russross/blackfriday/v2 v2.0.1 // indirect
 	github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
 	go.dedis.ch/fixbuf v1.0.3 // indirect
-	golang.org/x/crypto v0.7.0 // indirect
-	golang.org/x/net v0.8.0 // indirect
+	golang.org/x/crypto v0.14.0 // indirect
+	golang.org/x/net v0.16.0 // indirect
 	golang.org/x/oauth2 v0.6.0 // indirect
-	golang.org/x/sys v0.6.0 // indirect
-	golang.org/x/text v0.8.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/protobuf v1.29.1 // indirect
 	gopkg.in/square/go-jose.v2 v2.6.0 // indirect

be1-go/go.sum

@@ -21,13 +21,16 @@ github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
 github.com/golang-jwt/jwt/v4 v4.5.0 h1:7cYmW1XlMY7h7ii7UhUyChSgS5wUJEnm9uZVTGqOWzg=
 github.com/golang-jwt/jwt/v4 v4.5.0/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
 github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/schema v1.2.0 h1:YufUaxZYCKGFuAq3c96BOhjgd5nmXiOY9NGzF247Tsc=
@@ -91,8 +94,8 @@ golang.org/x/crypto v0.0.0-20190123085648-057139ce5d2b/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A=
-golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU=
+golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/exp v0.0.0-20230321023759-10a507213a29 h1:ooxPy7fPvB4kwsA2h+iBNHkAbp/4JxTSwCmvdjEYmug=
 golang.org/x/exp v0.0.0-20230321023759-10a507213a29/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -102,15 +105,15 @@ golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
-golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/net v0.16.0 h1:7eBu7KsSvFDtSXUIDbh3aqlK4DPsZ1rByC8PFfBThos=
+golang.org/x/net v0.16.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/oauth2 v0.6.0 h1:Lh8GPgSKBfWSwFvtuWOfeI3aAAnbXTSutYxJiOJFgIw=
 golang.org/x/oauth2 v0.6.0/go.mod h1:ycmewcwgD4Rpr3eZJLSB4Kyyljb3qDh40vJ8STE5HKw=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
-golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.4.0 h1:zxkM55ReGkDlKSM+Fu41A+zmbZuaPVbGMzvvdUPznYQ=
+golang.org/x/sync v0.4.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20190124100055-b90733256f2e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -120,14 +123,14 @@ golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
-golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68=
-golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
@@ -146,6 +149,7 @@ google.golang.org/protobuf v1.29.1 h1:7QBf+IK2gx70Ap/hDsOmam3GE0v9HicjfEdAxE62Uo
 google.golang.org/protobuf v1.29.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/square/go-jose.v2 v2.6.0 h1:NGk74WTnPKBNUhNzQX7PYcTLUjoq7mzKk2OKbvwk2iI=
 gopkg.in/square/go-jose.v2 v2.6.0/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=
 gopkg.in/yaml.v2 v2.2.3 h1:fvjTMHxHEw/mxHbtzPi3JCcKXQRAnQTBRo6YCJSVHKI=

be1-go/hub/standard_hub/hub_state/Channels.go

@@ -0,0 +1,24 @@
+package hub_state
+
+import "popstellar/channel"
+
+// Channels stores channel ids with their corresponding channels
+type Channels struct {
+	ThreadSafeMap[string, channel.Channel]
+}
+
+// NewChannelsMap creates a new Channels structure
+func NewChannelsMap() Channels {
+	return Channels{
+		ThreadSafeMap: NewThreadSafeMap[string, channel.Channel](),
+	}
+}
+
+// ForEach iterates over all channels and applies the given function
+func (c *Channels) ForEach(f func(channel.Channel)) {
+	c.Lock()
+	defer c.Unlock()
+	for _, channel := range c.table {
+		f(channel)
+	}
+}

be1-go/hub/standard_hub/hub_state/MessageIds.go

@@ -0,0 +1,39 @@
+package hub_state
+
+import (
+	"golang.org/x/exp/slices"
+)
+
+// MessageIds stores a channel id with its corresponding message ids
+type MessageIds struct {
+	ThreadSafeMap[string, []string]
+}
+
+// NewMessageIdsMap creates a new MessageIds structure
+func NewMessageIdsMap() MessageIds {
+	return MessageIds{
+		ThreadSafeMap: NewThreadSafeMap[string, []string](),
+	}
+}
+
+// Add adds a message id to the slice of message ids of the channel
+func (i *MessageIds) Add(channel string, id string) {
+	i.Lock()
+	defer i.Unlock()
+	messageIds, channelStored := i.table[channel]
+	if !channelStored {
+		i.table[channel] = append(i.table[channel], id)
+		return
+	}
+	alreadyStoredId := slices.Contains(messageIds, id)
+	if !alreadyStoredId {
+		i.table[channel] = append(i.table[channel], id)
+	}
+}
+
+// AddAll adds a slice of message ids to the slice of message ids of the channel
+func (i *MessageIds) AddAll(channel string, ids []string) {
+	i.Lock()
+	defer i.Unlock()
+	i.table[channel] = append(i.table[channel], ids...)
+}

be1-go/hub/standard_hub/hub_state/Peers.go

@@ -0,0 +1,60 @@
+package hub_state
+
+import (
+	"popstellar/message/query/method"
+	"sync"
+
+	"golang.org/x/exp/maps"
+	"golang.org/x/exp/slices"
+)
+
+// Peers stores the peers' information
+type Peers struct {
+	sync.RWMutex
+	// peersInfo stores the info of the peers: public key, client and server endpoints associated with the socket ID
+	peersInfo map[string]method.ServerInfo
+	// peersGreeted stores the peers that were greeted by the socket ID
+	peersGreeted map[string]struct{}
+}
+
+// NewPeers creates a new Peers structure
+func NewPeers() Peers {
+	return Peers{
+		peersInfo:    make(map[string]method.ServerInfo),
+		peersGreeted: make(map[string]struct{}),
+	}
+}
+
+// AddPeerInfo adds a peer's info to the table
+func (p *Peers) AddPeerInfo(socketId string, info method.ServerInfo) {
+	p.Lock()
+	defer p.Unlock()
+	p.peersInfo[socketId] = info
+}
+
+// AddPeerGreeted adds a peer's socket ID to the slice of peers greeted
+func (p *Peers) AddPeerGreeted(socketId string) {
+	p.Lock()
+	defer p.Unlock()
+	p.peersGreeted[socketId] = struct{}{}
+}
+
+// GetAllPeersInfo returns a copy of the peers' info slice
+func (p *Peers) GetAllPeersInfo() []method.ServerInfo {
+	p.RLock()
+	defer p.RUnlock()
+	peersInfo := make([]method.ServerInfo, 0, len(p.peersInfo))
+	for _, info := range p.peersInfo {
+		if !slices.Contains(peersInfo, info) {
+			peersInfo = append(peersInfo, info)
+		}
+	}
+	return peersInfo
+}
+
+// IsPeerGreeted returns true if the peer was greeted, otherwise it returns false
+func (p *Peers) IsPeerGreeted(socketId string) bool {
+	p.RLock()
+	defer p.RUnlock()
+	return slices.Contains(maps.Keys(p.peersGreeted), socketId)
+}

be1-go/hub/standard_hub/hub_state/Queries.go

@@ -0,0 +1,78 @@
+package hub_state
+
+import (
+	"popstellar/message/query/method"
+	"sync"
+
+	"golang.org/x/xerrors"
+)
+
+// Queries let the hub remember all queries that it sent to other servers
+type Queries struct {
+	sync.Mutex
+	// state stores the ID of the server's queries and their state. False for a
+	// query not yet answered, else true.
+	state map[int]bool
+	// getMessagesByIdQueries stores the server's getMessagesByIds queries by their ID.
+	getMessagesByIdQueries map[int]method.GetMessagesById
+	// nextID store the ID of the next query
+	nextID int
+}
+
+// NewQueries creates a new queries struct
+func NewQueries() Queries {
+	return Queries{
+		state:                  make(map[int]bool),
+		getMessagesByIdQueries: make(map[int]method.GetMessagesById),
+	}
+}
+
+// GetQueryState returns a given query's state
+func (q *Queries) GetQueryState(id int) (bool, error) {
+	q.Lock()
+	defer q.Unlock()
+
+	state, ok := q.state[id]
+	if !ok {
+		return false, xerrors.Errorf("query with id %d not found", id)
+	}
+	return state, nil
+}
+
+// GetNextID returns the next query ID
+func (q *Queries) GetNextID() int {
+	q.Lock()
+	defer q.Unlock()
+
+	id := q.nextID
+	q.nextID++
+	return id
+}
+
+// SetQueryReceived sets the state of the query with the given ID as received
+func (q *Queries) SetQueryReceived(id int) error {
+	q.Lock()
+	defer q.Unlock()
+
+	currentState, ok := q.state[id]
+
+	if !ok {
+		return xerrors.Errorf("query with id %d not found", id)
+	}
+
+	if currentState {
+		return xerrors.Errorf("query with id %d already answered", id)
+	}
+
+	q.state[id] = true
+	return nil
+}
+
+// AddQuery adds the given query to the table
+func (q *Queries) AddQuery(id int, query method.GetMessagesById) {
+	q.Lock()
+	defer q.Unlock()
+
+	q.getMessagesByIdQueries[id] = query
+	q.state[id] = false
+}