Function waitForWorker

controlplane/worker_mgr.go:516–561

waitForWorker polls for the worker socket and creates a Flight SQL client.

(socketPath, bearerToken string, timeout time.Duration)

	p.mu.Lock()
	p.workers[id] = w
	workerCount := len(p.workers)
	p.mu.Unlock()
	observeControlPlaneWorkers(workerCount)

	slog.Info("Worker spawned.", "id", id, "pid", cmd.Process.Pid, "socket", socketPath)
	return nil
}

// waitForWorker polls for the worker socket and creates a Flight SQL client.
func waitForWorker(socketPath, bearerToken string, timeout time.Duration) (*flightsql.Client, error) {
	deadline := time.Now().Add(timeout)
	var lastErr error
	attempts := 0

	for time.Now().Before(deadline) {
		if _, err := os.Stat(socketPath); err == nil {
			// Socket exists, try to connect
			addr := "unix://" + socketPath
			var dialOpts []grpc.DialOption
			dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
			dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(
				grpc.MaxCallRecvMsgSize(flightclient.MaxGRPCMessageSize),
				grpc.MaxCallSendMsgSize(flightclient.MaxGRPCMessageSize),
			))

			if bearerToken != "" {
				dialOpts = append(dialOpts, grpc.WithPerRPCCredentials(&workerBearerCreds{token: bearerToken}))
			}

			client, err := flightsql.NewClient(addr, nil, nil, dialOpts...)
			if err == nil {
				// Verify with a health check
				ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
				_, err = doHealthCheck(ctx, client)
				cancel()
				if err == nil {
					return client, nil
				}
				lastErr = err
				_ = client.Close()
			} else {
				lastErr = fmt.Errorf("grpc dial: %w", err)
			}
			attempts++
			if attempts <= 3 || attempts%10 == 0 {
				slog.Debug("waitForWorker health check attempt failed.", "socket", socketPath, "attempt", attempts, "error", lastErr)
			}
		} else {
			lastErr = err