Long-polling & leasing

The agent runs inside your network and opens no inbound ports. So the control plane can’t push work to it — the agent has to ask. Naive polling (“any jobs? …any jobs now? …now?”) is wasteful and laggy. Setu uses HTTP long-polling: the agent makes one request that the server holds open until a job appears or a timeout elapses. Step through it:

// agent: one call, held open by the server
url := c.baseURL + "/agent/v1/jobs/next?wait=" + itoa(waitSec)
// transport timeout = wait + 15s headroom so the
// server's 204 always wins the race, not the client
cl := &http.Client{Timeout: time.Duration(waitSec)*time.Second + 15*time.Second}

Why long-poll instead of WebSockets or a queue?

• Outbound-only. A plain HTTPS request traverses NAT, proxies, and corporate firewalls with zero config. A WebSocket also would, but adds a stateful connection to manage, scale, and reconnect.
• Stateless server. Each poll is an independent request; any apiserver instance can answer. No sticky sessions, no connection registry.
• Backpressure for free. The agent only asks for the next job when it’s ready for one. There’s no inbound flood to absorb.
• Batch-shaped work. Migration jobs are coarse (seconds-to-minutes), so a ~1s worst-case pickup latency is irrelevant — long-poll’s tradeoff is exactly right here.

The request: a bounded wait

The agent asks for the next job and states how long it’s willing to wait (wait seconds). Crucially, the HTTP client timeout is set longer than the server’s wait window, so the server’s own “no job” reply always wins the race — the client never times out a request the server was about to answer.

func (c *Client) JobsNext(ctx context.Context, waitSec int) (*Job, error) {
	url := c.baseURL + "/agent/v1/jobs/next?wait=" + strconv.Itoa(waitSec)
	req, err := c.newAuthed(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	// Give the transport headroom beyond the server-side long-poll window,
	// so the server's 204 (no job) always arrives before the client times out.
	cl := &http.Client{Timeout: time.Duration(waitSec)*time.Second + 15*time.Second}
	resp, err := cl.Do(req)
	// … 200 → decode & return the leased job; 204 → (nil, nil), re-poll; else error
}

The server hold loop

On the server, /agent/v1/jobs/next doesn’t block on a condition variable — it tries to lease in a loop, sleeping a short interval between attempts until either a job is leased or the caller’s wait window is spent (then 204). It honours request-context cancellation, so a client disconnect or a server shutdown ends the loop immediately.

func (s *Server) agentJobNext(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	agentID := currentAgent(ctx)

	wait := time.Duration(0)
	if v := r.URL.Query().Get("wait"); v != "" {
		if secs, err := strconv.Atoi(v); err == nil && secs > 0 {
			wait = time.Duration(secs) * time.Second
		}
	}
	if wait > jobMaxWait { // capped so a stuck agent can't hold a request forever
		wait = jobMaxWait
	}
	deadline := time.Now().Add(wait)

	for {
		job, err := s.q.LeaseNextJob(ctx, sqlc.LeaseNextJobParams{
			AgentID:      agentID,
			LeaseSeconds: int32(jobLeaseTTL / time.Second),
		})
		if err == nil {
			writeJSON(w, http.StatusOK, toAgentJobDTO(job)) // 200 + leased job
			return
		}
		if !errors.Is(err, pgx.ErrNoRows) {
			s.fail(w, http.StatusInternalServerError, err)
			return
		}
		if !time.Now().Before(deadline) { // window spent → no job
			w.WriteHeader(http.StatusNoContent) // 204
			return
		}
		select {
		case <-ctx.Done(): // client hung up / server shutting down
			return
		case <-time.After(jobPollInterval): // ~1s, then retry the lease
		}
	}
}

Atomic leasing — the concurrency gate

The single most important line is the lease. Multiple apiserver instances (and multiple polls) can run LeaseNextJob concurrently; correctness demands that a job is handed to exactly one agent poll. That guarantee is pushed entirely into one SQL statement: select the oldest pending job FOR UPDATE SKIP LOCKED and flip it to leased in the same transaction, RETURNING the row. Concurrent callers skip the locked row and either grab a different job or get ErrNoRows — never a double-lease.

-- name: LeaseNextJob :one
UPDATE agent_job SET
  status = 'leased',
  leased_at = now(),
  lease_expires_at = now() + make_interval(secs => sqlc.arg('lease_seconds')::int)
WHERE id = (
  SELECT id FROM agent_job
  WHERE agent_id = sqlc.arg('agent_id') AND status = 'pending'
  ORDER BY created_at
  FOR UPDATE SKIP LOCKED   -- ← the concurrency gate: no two pollers take the same row
  LIMIT 1
)
RETURNING *;

Heartbeat & the reaper

A lease is a lie unless it’s enforced. While the agent works a job it POSTs /heartbeat to push lease_expires_at forward. If the agent crashes mid-job, it stops heartbeating; a background reaper finds jobs whose lease has expired and returns them to pending, so another poll can pick them up. Work is never silently lost.

// Any job still 'leased' past its lease_expires_at is assumed abandoned
// (agent crash / network partition) and requeued to 'pending'.
RequeueExpiredJobs(ctx) // UPDATE agent_job SET status='pending', leased_at=NULL
                        // WHERE status='leased' AND lease_expires_at < now()

Backoff on transport errors

Distinct from the poll interval: when a poll fails (the control plane is down, DNS hiccup, a cold-started free-tier host returning 502), the agent backs off exponentially between attempts — bounded — so an outage doesn’t become a tight retry storm.

backoff := backoffMin
for {
	job, err := c.JobsNext(ctx, jobWaitSeconds)
	if err != nil {
		log.Printf("poll error: %v (retrying in %s)", err, backoff)
		time.Sleep(backoff)
		backoff = min(backoff*2, backoffMax) // exponential, capped
		continue
	}
	backoff = backoffMin // reset on success
	if job == nil {
		continue // 204: no job in the window, just poll again
	}
	dispatch(ctx, c, cfg, job) // test | extract | load | discover | preview
}