Add a new test and explain logic

Signed-off-by: Paschalis Tsilias <[email protected]>

client/internal/httpsender_test.go

@@ -172,3 +172,69 @@ func TestHTTPSenderRetryForFailedRequests(t *testing.T) {
 	cancel()
 	srv.Close()
 }
+
+func TestPackageUpdatesInParallel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	localPackageState := NewInMemPackagesStore()
+	sender := NewHTTPSender(&sharedinternal.NopLogger{})
+
+	var messages atomic.Int32
+	var mut sync.Mutex
+	sender.callbacks = types.CallbacksStruct{
+		OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
+			err := msg.PackageSyncer.Sync(ctx)
+			assert.NoError(t, err)
+			messages.Add(1)
+		},
+	}
+
+	// Set the RequestInstanceUid flag on the tracked state to request the server for a new ID to use.
+	clientSyncedState := &ClientSyncedState{}
+	capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsPackages
+	sender.receiveProcessor = newReceivedProcessor(&sharedinternal.NopLogger{}, sender.callbacks, sender, clientSyncedState, localPackageState, capabilities, &mut)
+
+	go func() {
+		sender.receiveProcessor.ProcessReceivedMessage(ctx,
+			&protobufs.ServerToAgent{
+				PackagesAvailable: &protobufs.PackagesAvailable{
+					Packages: map[string]*protobufs.PackageAvailable{
+						"package1": {
+							Type:    protobufs.PackageType_PackageType_TopLevel,
+							Version: "1.0.0",
+							File: &protobufs.DownloadableFile{
+								DownloadUrl: "foo",
+								ContentHash: []byte{4, 5},
+							},
+							Hash: []byte{1, 2, 3},
+						},
+					},
+					AllPackagesHash: []byte{1, 2, 3, 4, 5},
+				},
+			})
+	}()
+	go func() {
+		sender.receiveProcessor.ProcessReceivedMessage(ctx,
+			&protobufs.ServerToAgent{
+				PackagesAvailable: &protobufs.PackagesAvailable{
+					Packages: map[string]*protobufs.PackageAvailable{
+						"package22": {
+							Type:    protobufs.PackageType_PackageType_TopLevel,
+							Version: "1.0.0",
+							File: &protobufs.DownloadableFile{
+								DownloadUrl: "bar",
+								ContentHash: []byte{4, 5},
+							},
+							Hash: []byte{1, 2, 3},
+						},
+					},
+					AllPackagesHash: []byte{1, 2, 3, 4, 5},
+				},
+			})
+	}()
+
+	assert.Eventually(t, func() bool {
+		return messages.Load() == 2
+	}, 2*time.Second, 100*time.Millisecond, "both messages must have been processed successfully")
+
+	cancel()
+}

client/internal/packagessyncer.go

@@ -53,6 +53,8 @@ func (s *packagesSyncer) Sync(ctx context.Context) error {
 	}()
 
 	// Prepare package statuses.
+	// Grab a lock to make sure that package statuses are not overriden by
+	// another call to Sync running in parallel.
 	s.mut.Lock()
 	if err := s.initStatuses(); err != nil {
 		return err
@@ -62,7 +64,8 @@ func (s *packagesSyncer) Sync(ctx context.Context) error {
 		return err
 	}
 
-	// Now do the actual syncing in the background.
+	// Now do the actual syncing in the background and release the lock from
+	// inside of the goroutine.
 	go s.doSync(ctx)
 
 	return nil
@@ -104,6 +107,8 @@ func (s *packagesSyncer) initStatuses() error {
 
 // doSync performs the actual syncing process.
 func (s *packagesSyncer) doSync(ctx context.Context) {
+	// Once doSync returns  in a separate goroutine, make sure to release the
+	// mutex so that a new syncing process can take place.
 	defer s.mut.Unlock()
 
 	hash, err := s.localState.AllPackagesHash()