diff --git a/internal/sync/sync.go b/internal/sync/sync.go index f9828e8..6d63b8b 100644 --- a/internal/sync/sync.go +++ b/internal/sync/sync.go @@ -14,6 +14,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" "time" "github.com/mrhid6/vantage/agent/internal/config" @@ -169,6 +170,16 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error { log.Println("command stream connected") + // grpc streams are not safe for concurrent Send; RunStep results are sent + // from per-command goroutines, so all sends on this stream must go through + // this mutex-protected helper. + var sendMu sync.Mutex + send := func(msg *pb.AgentMessage) error { + sendMu.Lock() + defer sendMu.Unlock() + return stream.Send(msg) + } + for { cmd, err := stream.Recv() if err != nil { @@ -188,13 +199,15 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error { go handleApplyUpdates(cfg, cmd) } if cmd.RunStep != nil { - res := agentexec.RunStep(cmd.RunStep) - res.CommandId = cmd.CommandId - _ = stream.Send(&pb.AgentMessage{ - ServerId: cfg.ServerID, - AgentToken: cfg.AgentToken, - StepResult: res, - }) + go func(rc *pb.RunStepCmd, cid string) { + res := agentexec.RunStep(rc) + res.CommandId = cid + _ = send(&pb.AgentMessage{ + ServerId: cfg.ServerID, + AgentToken: cfg.AgentToken, + StepResult: res, + }) + }(cmd.RunStep, cmd.CommandId) continue } }