diff --git a/apps/swarm/main.c b/apps/swarm/main.c
index ae39c64..258ce53 100644
--- a/apps/swarm/main.c
+++ b/apps/swarm/main.c
@@ -39,6 +39,8 @@
 #define REQUEST_TIMEOUT 15.0
 #define CONNECT_TIMEOUT_MS 5000
 #define EXT_RESERVED 0x0000000000100000ULL
+#define DEFAULT_TARGET_PEERS 80
+#define MAX_TARGET_PEERS 512
 
 typedef struct {
     uint32_t piece, begin, length;
@@ -56,12 +58,16 @@ typedef struct {
     char     name[40];
     uint8_t  peer_id[20];
     bool     have_peer_id;
+    bool     connecting;
     uint8_t *rbuf; size_t rcap, rlen;
     bool     hs_done, peer_choking;
     naut_bitfield have;
     req_t   *inflight; size_t nflight, cflight;
     uint64_t blocks_received;
     uint64_t bytes_received;
+    uint64_t rate_last_bytes;
+    double   rate_last_at;
+    double   download_rate;
     naut_ext_handshake extensions;
     uint64_t pex_received;
     naut_pipeline pipeline;
@@ -70,6 +76,15 @@ typedef struct {
 
 static double now(void) { struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t); return t.tv_sec + t.tv_nsec*1e-9; }
 
+static uint32_t target_peer_count(void) {
+    const char *env = getenv("NAUT_TARGET_PEERS");
+    if (!env || !*env) return DEFAULT_TARGET_PEERS;
+    char *end = NULL;
+    unsigned long value = strtoul(env, &end, 10);
+    if (!end || *end || value == 0) return DEFAULT_TARGET_PEERS;
+    return (uint32_t)NAUT_MIN(value, MAX_TARGET_PEERS);
+}
+
 static void random_bytes(uint8_t *output, size_t length) {
     int fd = open("/dev/urandom", O_RDONLY);
     size_t offset = 0;
@@ -161,14 +176,33 @@ static void peer_flags(const peer_t *peer, char out[16]) {
     out[n] = 0;
 }
 
+static void peer_update_rate(peer_t *peer, double sampled_at) {
+    if (!peer || peer->dead) return;
+    if (peer->rate_last_at <= 0) {
+        peer->rate_last_at = sampled_at;
+        peer->rate_last_bytes = peer->bytes_received;
+        return;
+    }
+    double dt = sampled_at - peer->rate_last_at;
+    if (dt < 0.25) return;
+    uint64_t delta = peer->bytes_received - peer->rate_last_bytes;
+    double rate = (double)delta / dt;
+    peer->download_rate = peer->download_rate <= 0.0
+        ? rate : peer->download_rate * 0.7 + rate * 0.3;
+    peer->rate_last_at = sampled_at;
+    peer->rate_last_bytes = peer->bytes_received;
+}
+
 static void snapshot_peer_stats(naut_swarm_stats *stats,
-                                const peer_t *peers, int npeers,
+                                peer_t *peers, int npeers,
                                 uint32_t total_pieces) {
     if (!stats || !peers || npeers <= 0) return;
+    double sampled_at = now();
     for (int i = 0; i < npeers &&
                     stats->peer_count < NAUT_SWARM_MAX_PEER_STATS; i++) {
-        const peer_t *peer = &peers[i];
+        peer_t *peer = &peers[i];
         if (peer->dead || !peer->hs_done) continue;
+        peer_update_rate(peer, sampled_at);
         naut_swarm_peer_stats *out =
             &stats->peer_stats[stats->peer_count++];
         snprintf(out->ip, sizeof out->ip, "%u.%u.%u.%u",
@@ -184,7 +218,7 @@ static void snapshot_peer_stats(naut_swarm_stats *stats,
         if (ratio > 1.0) ratio = 1.0;
         out->progress = ratio;
         out->relevance = ratio;
-        out->dlspeed = peer->pipeline.bytes_per_second;
+        out->dlspeed = peer->download_rate;
         out->upspeed = 0.0;
         out->downloaded = peer->bytes_received;
         out->uploaded = 0;
@@ -196,7 +230,7 @@ static void report_progress(const naut_swarm_config *config,
                             const naut_metainfo *metainfo,
                             uint32_t peers_total, uint32_t peers_connecting,
                             uint32_t peers_active, uint32_t peers_failed,
-                            const peer_t *peers, int npeers,
+                            peer_t *peers, int npeers,
                             double started_at) {
     if (!config->on_progress) return;
     naut_swarm_stats stats = {
@@ -475,6 +509,7 @@ static void peer_drop(naut_download *d, peer_t *p) {
     peer_release(d, p);
     if (p->fd >= 0) close(p->fd);
     p->fd = -1;
+    p->connecting = false;
     p->dead = true;
 }
 
@@ -554,6 +589,7 @@ static bool peer_start(naut_download *download, const naut_metainfo *metainfo,
     peer->fd = fd;
     peer->addr = endpoint->addr;
     snprintf(peer->name, sizeof peer->name, "%s", endpoint->name);
+    peer->connecting = false;
     peer->peer_choking = true;
     naut_pipeline_init(&peer->pipeline, NAUT_BLOCK, 4, 1024, 32);
     peer->rcap = 1 << 18;
@@ -589,6 +625,113 @@ static bool peer_start(naut_download *download, const naut_metainfo *metainfo,
     return true;
 }
 
+static bool ensure_peer_capacity(peer_t **peers, struct pollfd **pfd,
+                                 int **idx_map, int *capacity, int want) {
+    if (want <= *capacity) return true;
+    int next = *capacity > 0 ? *capacity : 1;
+    while (next < want) next *= 2;
+    peer_t *next_peers = realloc(*peers, (size_t)next * sizeof(*next_peers));
+    if (!next_peers) return false;
+    for (int i = *capacity; i < next; i++) {
+        memset(&next_peers[i], 0, sizeof(next_peers[i]));
+        next_peers[i].fd = -1;
+        next_peers[i].dead = true;
+    }
+    struct pollfd *next_pfd =
+        realloc(*pfd, ((size_t)next + 1) * sizeof(*next_pfd));
+    if (!next_pfd) {
+        *peers = next_peers;
+        return false;
+    }
+    int *next_idx = realloc(*idx_map, ((size_t)next + 1) * sizeof(*next_idx));
+    if (!next_idx) {
+        *peers = next_peers;
+        *pfd = next_pfd;
+        return false;
+    }
+    *peers = next_peers;
+    *pfd = next_pfd;
+    *idx_map = next_idx;
+    *capacity = next;
+    return true;
+}
+
+static bool peer_addr_seen(const peer_t *peers, int npeers,
+                           const naut_peer_addr *addr);
+
+static uint32_t peer_count_active(const peer_t *peers, int npeers,
+                                  bool include_connecting) {
+    uint32_t count = 0;
+    for (int i = 0; i < npeers; i++) {
+        if (peers[i].dead) continue;
+        if (peers[i].connecting && !include_connecting) continue;
+        count++;
+    }
+    return count;
+}
+
+static bool connect_one_candidate(const naut_swarm_config *config,
+                                  naut_download *download,
+                                  const naut_metainfo *metainfo,
+                                  const uint8_t peerid[20],
+                                  peer_t **peers, int *npeers,
+                                  int *peer_capacity,
+                                  struct pollfd **pfd, int **idx_map,
+                                  endpoint_t endpoint,
+                                  uint32_t *failed) {
+    if (!ensure_peer_capacity(peers, pfd, idx_map, peer_capacity,
+                              *npeers + 1))
+        return false;
+    peer_t *peer = &(*peers)[*npeers];
+    memset(peer, 0, sizeof(*peer));
+    peer->fd = -1;
+    peer->addr = endpoint.addr;
+    snprintf(peer->name, sizeof peer->name, "%s", endpoint.name);
+    bool connected = false;
+    int fd = connect_start(&endpoint, &connected);
+    if (fd < 0) {
+        if (failed) (*failed)++;
+        return true;
+    }
+    peer->fd = fd;
+    (*npeers)++;
+    if (connected) {
+        if (!connect_finish(fd) ||
+            !peer_start(download, metainfo, peerid, &endpoint, peer, fd)) {
+            if (peer->fd >= 0) close(peer->fd);
+            peer->fd = -1;
+            peer->dead = true;
+            if (failed) (*failed)++;
+            return true;
+        }
+        emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0, peer->name, NULL);
+        return true;
+    }
+    peer->connecting = true;
+    return true;
+}
+
+static bool connect_pending_peers(const naut_swarm_config *config,
+                                  naut_download *download,
+                                  const naut_metainfo *metainfo,
+                                  const uint8_t peerid[20],
+                                  peer_t **peers, int *npeers,
+                                  int *peer_capacity,
+                                  struct pollfd **pfd, int **idx_map,
+                                  endpoint_t *pending, size_t *npending,
+                                  uint32_t target, uint32_t *failed) {
+    while (*npending > 0 &&
+           peer_count_active(*peers, *npeers, true) < target) {
+        endpoint_t endpoint = pending[--(*npending)];
+        if (peer_addr_seen(*peers, *npeers, &endpoint.addr)) continue;
+        if (!connect_one_candidate(config, download, metainfo, peerid,
+                                   peers, npeers, peer_capacity, pfd, idx_map,
+                                   endpoint, failed))
+            return false;
+    }
+    return true;
+}
+
 /* process all complete messages currently buffered for peer p */
 static void cancel_block(naut_download *d, peer_t *peers, int npeers,
                          peer_t *source, uint32_t piece, uint32_t begin) {
@@ -646,8 +789,19 @@ static void merge_bitfield(naut_download *d, const naut_metainfo *mi,
     naut_bitfield_free(&incoming);
 }
 
+static bool peer_addr_seen(const peer_t *peers, int npeers,
+                           const naut_peer_addr *addr) {
+    for (int i = 0; i < npeers; i++)
+        if (peers[i].addr.port == addr->port &&
+            memcmp(peers[i].addr.ip, addr->ip, sizeof addr->ip) == 0)
+            return true;
+    return false;
+}
+
 static naut_err peer_process(naut_download *d, const naut_metainfo *mi,
-                             peer_t *peers, int npeers, peer_t *p) {
+                             peer_t *peers, int npeers, peer_t *p,
+                             endpoint_t **pending, size_t *npending,
+                             size_t *pending_cap) {
     size_t pos = 0;
     if (!p->hs_done) {
         if (p->rlen < NAUT_HANDSHAKE_LEN) return NAUT_OK;
@@ -693,28 +847,33 @@ static naut_err peer_process(naut_download *d, const naut_metainfo *mi,
                     peer_drop(d, p);
                     goto parsed;
                 }
-            } else if (m.payload[0] == NAUT_EXT_UT_PEX) {
+            } else if (p->extensions.ut_pex &&
+                       m.payload[0] == p->extensions.ut_pex) {
                 naut_pex_msg pex;
                 if (naut_pex_parse(m.payload + 1, m.payload_len - 1,
                                    &pex) != NAUT_OK) {
                     peer_drop(d, p);
                     goto parsed;
                 }
-                p->pex_received += pex.num_added;
+                for (size_t i = 0; i < pex.num_added; i++)
+                    if (!peer_addr_seen(peers, npeers, &pex.added[i]) &&
+                        endpoint_add(pending, npending, pending_cap,
+                                     &pex.added[i])) {
+                        p->pex_received++;
+                    }
                 naut_pex_free(&pex);
             }
             break;
         case NAUT_MSG_PIECE: {
             req_t request;
             bool expected = peer_del_inflight(p, m.index, m.begin, &request);
-            if (expected) {
-                naut_pipeline_on_block(&p->pipeline, request.length,
-                                       request.sent_at, now());
-                naut_download_unrequest(d, m.index, m.begin);
-                if (request.length != m.payload_len) {
-                    peer_drop(d, p);
-                    goto parsed;
-                }
+            if (!expected) break;
+            naut_pipeline_on_block(&p->pipeline, request.length,
+                                   request.sent_at, now());
+            naut_download_unrequest(d, m.index, m.begin);
+            if (request.length != m.payload_len) {
+                peer_drop(d, p);
+                goto parsed;
             }
             bool pdone = false;
             naut_err e = naut_download_on_block(d, m.index, m.begin, m.payload,
@@ -724,8 +883,7 @@ static naut_err peer_process(naut_download *d, const naut_metainfo *mi,
                 p->bytes_received += m.payload_len;
                 cancel_block(d, peers, npeers, p, m.index, m.begin);
             } else if (e == NAUT_ERR_PROTO) {
-                if (expected) cancel_piece(d, peers, npeers, m.index);
-                else peer_drop(d, p);
+                cancel_piece(d, peers, npeers, m.index);
             } else {
                 return e;
             }
@@ -775,6 +933,7 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
 
     endpoint_t *endpoints = NULL;
     size_t neps = 0, epcap = 0;
+    uint32_t target_peers = target_peer_count();
     if (config->num_peers > 0) {
         for (size_t i = 0; i < config->num_peers; i++) {
             naut_peer_addr addr;
@@ -803,7 +962,7 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
         if (!discover_trackers(hash, total, trackers, num_trackers,
                                tracker_tiers, peerid, &endpoints, &neps,
                                &epcap) ||
-            (neps == 0 &&
+            (neps < target_peers &&
              !discover_dht(hash, &endpoints, &neps, &epcap))) {
             NAUT_ERROR("out of memory collecting discovered peers");
             naut_metainfo_free(&mi);
@@ -893,13 +1052,32 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
     }
     naut_download_set_worker_pool(d, workers);
 
-    int npeers = (int)neps;
-    peer_t *peers = calloc(neps, sizeof(*peers));
-    struct pollfd *pfd = calloc(neps + 1, sizeof(*pfd));
-    int *idx_map = calloc(neps + 1, sizeof(*idx_map));
+    endpoint_t *pending = NULL;
+    size_t npending = 0, pending_cap = 0;
+    size_t initial_endpoints = neps;
+    if (config->num_peers == 0 && initial_endpoints > target_peers)
+        initial_endpoints = target_peers;
+    for (size_t i = neps; i > initial_endpoints; i--) {
+        if (!endpoint_add(&pending, &npending, &pending_cap,
+                          &endpoints[i - 1].addr)) {
+            NAUT_ERROR("out of memory queueing discovered peers");
+            naut_worker_pool_destroy(workers);
+            naut_download_destroy(d);
+            naut_storage_close(st);
+            naut_metainfo_free(&mi);
+            free(endpoints);
+            return NAUT_ERR_NOMEM;
+        }
+    }
+
+    int npeers = (int)initial_endpoints;
+    int peer_capacity = (int)initial_endpoints;
+    peer_t *peers = calloc(initial_endpoints, sizeof(*peers));
+    struct pollfd *pfd = calloc(initial_endpoints + 1, sizeof(*pfd));
+    int *idx_map = calloc(initial_endpoints + 1, sizeof(*idx_map));
     if (!peers || !pfd || !idx_map) {
         NAUT_ERROR("out of memory creating swarm");
-        free(peers); free(pfd); free(idx_map);
+        free(peers); free(pfd); free(idx_map); free(pending);
         naut_worker_pool_destroy(workers);
         naut_download_destroy(d);
         naut_storage_close(st);
@@ -1010,19 +1188,32 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
             cancelled = true;
             break;
         }
+        if (!connect_pending_peers(config, d, &mi, peerid, &peers, &npeers,
+                                   &peer_capacity, &pfd, &idx_map,
+                                   pending, &npending, target_peers, &failed)) {
+            run_error = NAUT_ERR_NOMEM;
+            break;
+        }
         int nf = 0;
+        int connecting_peers = 0;
+        int live_peers = 0;
         for (int i = 0; i < npeers; i++) {
             if (peers[i].dead) continue;
-            pfd[nf].fd = peers[i].fd; pfd[nf].events = POLLIN; pfd[nf].revents = 0;
+            pfd[nf].fd = peers[i].fd;
+            pfd[nf].events = peers[i].connecting
+                ? (POLLOUT | POLLERR | POLLHUP | POLLNVAL)
+                : POLLIN;
+            pfd[nf].revents = 0;
             idx_map[nf] = i; nf++;
+            if (peers[i].connecting) connecting_peers++;
+            else live_peers++;
         }
-        int live_peers = nf;
         pfd[nf].fd = naut_worker_eventfd(workers);
         pfd[nf].events = POLLIN;
         pfd[nf].revents = 0;
         idx_map[nf] = -1;
         nf++;
-        if (live_peers == 0) {
+        if (live_peers == 0 && connecting_peers == 0) {
             uint32_t completed = 0;
             run_error = naut_download_poll(d, &completed);
             if (naut_download_complete(d)) break;
@@ -1031,9 +1222,10 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
             run_error = NAUT_ERR_IO;
             break;
         }
-        report_progress(config, d, &mi, (uint32_t)npeers, 0,
+        report_progress(config, d, &mi, (uint32_t)npeers,
+                        (uint32_t)connecting_peers,
                         (uint32_t)live_peers,
-                        (uint32_t)npeers - (uint32_t)live_peers,
+                        failed,
                         peers, npeers, t0);
         int r = poll(pfd, nf, 200);
         if (r < 0) {
@@ -1053,6 +1245,27 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
                 continue;
             }
             peer_t *p = &peers[idx_map[k]];
+            if (p->connecting) {
+                if (!(pfd[k].revents &
+                      (POLLOUT | POLLERR | POLLHUP | POLLNVAL)))
+                    continue;
+                endpoint_t endpoint = { .addr = p->addr };
+                for (size_t c = 0; c + 1 < sizeof endpoint.name; c++) {
+                    endpoint.name[c] = p->name[c];
+                    if (p->name[c] == 0) break;
+                }
+                endpoint.name[sizeof endpoint.name - 1] = 0;
+                if (!connect_finish(p->fd) ||
+                    !peer_start(d, &mi, peerid, &endpoint, p, p->fd)) {
+                    NAUT_WARN("connect %s failed", p->name);
+                    peer_drop(d, p);
+                    failed++;
+                    continue;
+                }
+                emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0,
+                           p->name, NULL);
+                continue;
+            }
             if (!(pfd[k].revents & (POLLIN | POLLHUP | POLLERR))) continue;
             if (p->rlen == p->rcap) {
                 size_t cap = p->rcap * 2;
@@ -1064,7 +1277,8 @@ naut_err naut_swarm_run(const naut_swarm_config *config) {
             ssize_t got = recv(p->fd, p->rbuf + p->rlen, p->rcap - p->rlen, 0);
             if (got <= 0) { peer_drop(d, p); continue; }
             p->rlen += (size_t)got;
-            run_error = peer_process(d, &mi, peers, npeers, p);
+            run_error = peer_process(d, &mi, peers, npeers, p,
+                                     &pending, &npending, &pending_cap);
             if (run_error != NAUT_OK) break;
             if (p->dead) peer_drop(d, p);
         }
@@ -1106,14 +1320,15 @@ done:
     ok = naut_download_complete(d);
     service_control(config, st);
     naut_storage_sync(st);
+    double elapsed = now() - t0;
     for (int i = 0; i < npeers; i++) {
         if (peers[i].blocks_received)
-            NAUT_INFO("peer %s delivered %llu blocks (pipeline %u, RTT %.1f ms, %.1f MB/s)",
+            NAUT_INFO("peer %s delivered %llu blocks (pipeline %u, RTT %.1f ms, avg %.1f MB/s)",
                       peers[i].name,
                       (unsigned long long)peers[i].blocks_received,
                       naut_pipeline_depth(&peers[i].pipeline),
                       peers[i].pipeline.rtt_seconds * 1000.0,
-                      peers[i].pipeline.bytes_per_second / 1e6);
+                      elapsed > 0 ? peers[i].bytes_received / elapsed / 1e6 : 0.0);
         if (peers[i].pex_received)
             NAUT_INFO("peer %s advertised %llu peers through PEX",
                       peers[i].name,
@@ -1123,6 +1338,7 @@ done:
         if (peers[i].have.words) naut_bitfield_free(&peers[i].have);
     }
     free(peers); free(pfd); free(idx_map);
+    free(pending);
     naut_worker_pool_destroy(workers);
     naut_download_destroy(d); naut_storage_close(st); naut_metainfo_free(&mi);
     if (ok) return NAUT_OK;
diff --git a/src/piece/piece.c b/src/piece/piece.c
index 1b365d3..80b2e0a 100644
--- a/src/piece/piece.c
+++ b/src/piece/piece.c
@@ -9,6 +9,7 @@
 #define BLK NAUT_BLOCK          /* 16 KiB */
 #define ENDGAME_BLOCKS 8        /* switch to endgame when this few remain */
 #define ENDGAME_COPIES 2        /* at most two peers race a missing block */
+#define PIECE_INFLIGHT_SOFT_CAP 64
 
 /* per-piece in-progress state, lazily allocated and freed on completion */
 typedef struct {
@@ -37,6 +38,7 @@ struct naut_download {
     pstate   **ps;          /* [num_pieces] in-progress state or NULL */
 
     uint32_t cur_piece;     /* sequential cursor for next_request() */
+    uint32_t pick_cursor;   /* rotating start point for rarest-first ties */
 
     uint64_t total_blocks, recv_blocks;
     uint32_t pieces_done;
@@ -205,6 +207,12 @@ static uint32_t first_unreq(const pstate *s) {
     return UINT32_MAX;
 }
 
+static uint32_t piece_inflight(const pstate *s) {
+    uint32_t n = 0;
+    for (uint32_t b = 0; b < s->nblocks; b++) n += s->req_count[b];
+    return n;
+}
+
 static bool hand_out(naut_download *d, uint32_t p, uint32_t b,
                      uint32_t *index, uint32_t *begin, uint32_t *length) {
     d->ps[p]->req_count[b]++;
@@ -218,15 +226,20 @@ bool naut_download_pick_for_peer(naut_download *d, const naut_bitfield *peer_hav
     d->endgame = (d->total_blocks - d->recv_blocks) <= ENDGAME_BLOCKS;
 
     /* pass 1: finish an in-progress piece the peer has (reduces fragmentation) */
-    for (uint32_t p = 0; p < d->num_pieces; p++) {
+    for (uint32_t n = 0; n < d->num_pieces; n++) {
+        uint32_t p = (d->pick_cursor + n) % d->num_pieces;
         if (naut_bitfield_test(&d->have, p) || !d->ps[p]) continue;
         if (p >= peer_have->nbits || !naut_bitfield_test(peer_have, p)) continue;
+        if (!d->endgame &&
+            piece_inflight(d->ps[p]) >= PIECE_INFLIGHT_SOFT_CAP)
+            continue;
         uint32_t b = first_unreq(d->ps[p]);
         if (b != UINT32_MAX) return hand_out(d, p, b, index, begin, length);
     }
     /* pass 2: start the rarest new piece the peer has */
     uint32_t best = UINT32_MAX, best_av = UINT32_MAX;
-    for (uint32_t p = 0; p < d->num_pieces; p++) {
+    for (uint32_t n = 0; n < d->num_pieces; n++) {
+        uint32_t p = (d->pick_cursor + n) % d->num_pieces;
         if (naut_bitfield_test(&d->have, p) || d->ps[p]) continue;
         if (p >= peer_have->nbits || !naut_bitfield_test(peer_have, p) ||
             d->avail[p] == 0) continue;
@@ -234,12 +247,14 @@ bool naut_download_pick_for_peer(naut_download *d, const naut_bitfield *peer_hav
     }
     if (best != UINT32_MAX) {
         if (!ensure_ps(d, best)) return false;
+        d->pick_cursor = (best + 1) % d->num_pieces;
         return hand_out(d, best, 0, index, begin, length);
     }
     /* pass 3: endgame — race each missing block on at most two distinct peers */
     if (d->endgame) {
         for (uint8_t copies = 1; copies < ENDGAME_COPIES; copies++) {
-            for (uint32_t p = 0; p < d->num_pieces; p++) {
+            for (uint32_t n = 0; n < d->num_pieces; n++) {
+                uint32_t p = (d->pick_cursor + n) % d->num_pieces;
                 if (naut_bitfield_test(&d->have, p) ||
                     p >= peer_have->nbits ||
                     !naut_bitfield_test(peer_have, p)) continue;