diff --git a/apisix/balancer.lua b/apisix/balancer.lua
index 1647bab2f64e..ff2f8fb8dcaf 100644
--- a/apisix/balancer.lua
+++ b/apisix/balancer.lua
@@ -174,9 +174,9 @@ local function create_server_picker(upstream, checker)
             return server_picker
         end
 
-        core.log.info("upstream nodes: ",
-                      core.json.delay_encode(up_nodes[up_nodes._priority_index[1]]))
-        local server_picker = picker.new(up_nodes[up_nodes._priority_index[1]], upstream)
+        local priority = up_nodes._priority_index[1]
+        core.log.info("upstream nodes: ", core.json.delay_encode(up_nodes[priority]))
+        local server_picker = picker.new(up_nodes[priority], upstream, priority)
         server_picker.addr_to_domain = addr_to_domain
         return server_picker
     end
@@ -248,7 +248,13 @@ local function pick_server(route, ctx)
     local up_conf = ctx.upstream_conf
 
     local nodes_count = #up_conf.nodes
-    if nodes_count == 1 then
+    -- least_conn counts the in-flight connections of every request it routes, so it
+    -- has to see them even while the upstream has a single node: those connections
+    -- are what tells a later scale out that the node is not empty. Skipping the
+    -- balancer here would leave it blind to everything routed before the second
+    -- node showed up, which is the state a k8s deployment or a discovery service
+    -- starts from. See #12217
+    if nodes_count == 1 and up_conf.type ~= "least_conn" then
         local node = up_conf.nodes[1]
         ctx.balancer_ip = node.host
         ctx.balancer_port = node.port
@@ -268,7 +274,10 @@ local function pick_server(route, ctx)
     ctx.balancer_try_count = (ctx.balancer_try_count or 0) + 1
     if ctx.balancer_try_count > 1 then
         if ctx.server_picker and ctx.server_picker.after_balance then
-            ctx.server_picker.after_balance(ctx, true)
+            -- remembering the server as tried is what keeps the next pick off it, so
+            -- only do it when there is another one to move to. With a single node the
+            -- retry has to land on it again, the way the fast path below always did
+            ctx.server_picker.after_balance(ctx, nodes_count > 1)
         end
 
         if checker then
@@ -331,6 +340,9 @@ local function pick_server(route, ctx)
         return nil, "failed to find valid upstream server, all upstream servers tried"
     end
     ctx.balancer_server = server
+    -- from here on the request holds a server, so the log phase must be able to
+    -- release it even if we bail out below
+    ctx.server_picker = server_picker
 
     local domain = server_picker.addr_to_domain[server]
     local res, err = lrucache_addr(server, nil, parse_addr, server)
@@ -347,7 +359,6 @@ local function pick_server(route, ctx)
     if is_http and ctx.var then
         ctx.var.upstream_unresolved_host = ctx.upstream_unresolved_host
     end
-    ctx.server_picker = server_picker
     res.upstream_host = parse_server_for_upstream_host(res, ctx.upstream_scheme)
 
     return res
diff --git a/apisix/balancer/chash.lua b/apisix/balancer/chash.lua
index f0e971a3626e..a8ab632fb0b4 100644
--- a/apisix/balancer/chash.lua
+++ b/apisix/balancer/chash.lua
@@ -123,6 +123,11 @@ function _M.new(up_nodes, upstream)
             return servers[id]
         end,
         after_balance = function (ctx, before_retry)
+            -- the release is what makes the request stop holding the server, so drop
+            -- the reference here instead of leaving every caller to remember
+            local server = ctx.balancer_server
+            ctx.balancer_server = nil
+
             if not before_retry then
                 if ctx.balancer_tried_servers then
                     core.tablepool.release("balancer_tried_servers", ctx.balancer_tried_servers)
@@ -132,11 +137,15 @@ function _M.new(up_nodes, upstream)
                 return nil
             end
 
+            if not server then
+                return nil
+            end
+
             if not ctx.balancer_tried_servers then
                 ctx.balancer_tried_servers = core.tablepool.fetch("balancer_tried_servers", 0, 2)
             end
 
-            ctx.balancer_tried_servers[ctx.balancer_server] = true
+            ctx.balancer_tried_servers[server] = true
             ctx.balancer_tried_servers_count = (ctx.balancer_tried_servers_count or 0) + 1
         end,
         before_retry_next_priority = function (ctx)
diff --git a/apisix/balancer/ewma.lua b/apisix/balancer/ewma.lua
index 0834397b9f3a..93707c400ad5 100644
--- a/apisix/balancer/ewma.lua
+++ b/apisix/balancer/ewma.lua
@@ -187,12 +187,21 @@ local function _ewma_find(ctx, up_nodes)
 end
 
 local function _ewma_after_balance(ctx, before_retry)
+    -- the release is what makes the request stop holding the server, so drop the
+    -- reference here instead of leaving every caller to remember
+    local server = ctx.balancer_server
+    ctx.balancer_server = nil
+
     if before_retry then
+        if not server then
+            return nil
+        end
+
         if not ctx.balancer_tried_servers then
             ctx.balancer_tried_servers = core.tablepool.fetch("balancer_tried_servers", 0, 2)
         end
 
-        ctx.balancer_tried_servers[ctx.balancer_server] = true
+        ctx.balancer_tried_servers[server] = true
         ctx.balancer_tried_servers_count = (ctx.balancer_tried_servers_count or 0) + 1
 
         return nil
diff --git a/apisix/balancer/least_conn.lua b/apisix/balancer/least_conn.lua
index 8923d1781d00..3aa008ffc3db 100644
--- a/apisix/balancer/least_conn.lua
+++ b/apisix/balancer/least_conn.lua
@@ -24,66 +24,184 @@ local pairs = pairs
 local _M = {}
 
 
+-- Per-worker balancing state, shared by every picker built for the same upstream.
+--
+-- A picker is cached by the upstream version, so it is thrown away whenever the
+-- upstream changes: scaling, a config update, or a health status flip. The
+-- requests that are still in flight keep the picker they were routed with, and
+-- release their server on that picker in the log phase. If the heap lived inside
+-- the picker, the rebuilt one would never see those releases: the connections
+-- established before the rebuild would be forgotten on creation and then, once
+-- they closed, decremented on a heap nobody reads anymore. Long-lived
+-- connections (WebSocket) would keep the load skewed on the original nodes and
+-- least_conn would degrade to round-robin. See #12217.
+--
+-- Keeping the heap and the in-flight counts here, keyed by something stable
+-- across scaling, gives every generation of pickers a single view of the load.
+--
+-- The values are weak, which is exactly the lifetime this state needs. A picker
+-- holds its state, an in-flight request holds the picker it was routed with, and
+-- the picker cache holds the current one - so a state survives for as long as
+-- anything can still release a connection into it. Once nothing references it
+-- there is no connection left to count, and dropping it costs nothing. That also
+-- means there is no size to tune and no eviction that could quietly forget a busy
+-- upstream, which is what an LRU would do here: it would rank states by how
+-- recently they were rebuilt, and a stable upstream holding many long-lived
+-- connections is precisely the one that is never rebuilt.
+local states = setmetatable({}, {__mode = "v"})
+
+
 local function least_score(a, b)
     return a.score < b.score
 end
 
 
-function _M.new(up_nodes, upstream)
-    local servers_heap = binaryHeap.minUnique(least_score)
+local function new_state()
+    return {
+        heap = binaryHeap.minUnique(least_score),
+        -- server -> in-flight connections, only holds positive counts
+        conns = {},
+        -- server -> true, mirrors the payloads currently in the heap
+        members = {},
+    }
+end
+
+
+local function update_score(state, server)
+    local info = state.heap:valueByPayload(server)
+    -- the server may have left the upstream while it still held connections
+    if not info then
+        return
+    end
+
+    info.score = (1 + (state.conns[server] or 0)) * info.effect_weight
+    state.heap:update(server, info)
+end
+
+
+-- Align the long-lived heap with the current node set, keeping the in-flight
+-- counts of the nodes that survive. A node that is added back later (scaled in
+-- again, or reported healthy again) gets its score restored from `conns`.
+local function sync_nodes(state, up_nodes)
+    local heap = state.heap
+
+    for server in pairs(state.members) do
+        if not up_nodes[server] then
+            heap:remove(server)
+            state.members[server] = nil
+        end
+    end
+
     for server, weight in pairs(up_nodes) do
-        local score = 1 / weight
-        -- Note: the argument order of insert is different from others
-        servers_heap:insert({
-            server = server,
-            effect_weight = 1 / weight,
-            score = score,
-        }, server)
+        local effect_weight = 1 / weight
+        local info = heap:valueByPayload(server)
+        if info then
+            info.effect_weight = effect_weight
+        else
+            -- Note: the argument order of insert is different from others
+            heap:insert({
+                server = server,
+                effect_weight = effect_weight,
+                score = effect_weight,
+            }, server)
+            state.members[server] = true
+        end
+        -- one place decides what a score is worth
+        update_score(state, server)
     end
+end
+
+
+function _M.new(up_nodes, upstream, priority)
+    -- resource_key identifies the upstream and is stable across node scaling, unlike
+    -- the picker version which changes whenever the nodes change. Do not fall back to
+    -- resource_id: it is a bare id, so a route and an upstream sharing one would land
+    -- on the same heap and evict each other's nodes
+    local up_key = upstream.resource_key
+    local state
+    if up_key and priority then
+        -- each priority level owns a disjoint node set, so it needs its own heap.
+        -- Do not default the priority: a caller that does not name one has a node
+        -- set we cannot place, and folding it into level 0 would let sync_nodes
+        -- evict that level's nodes from the heap it shares
+        local state_key = up_key .. "#" .. priority
+        state = states[state_key]
+        if not state then
+            state = new_state()
+            states[state_key] = state
+        end
+    else
+        -- no stable identity, fall back to a state private to this picker
+        state = new_state()
+    end
+
+    sync_nodes(state, up_nodes)
+
+    local servers_heap = state.heap
+    local conns = state.conns
 
     return {
         upstream = upstream,
         get = function (ctx)
+            local tried = ctx.balancer_tried_servers
             local server, info, err
-            if ctx.balancer_tried_servers then
-                local tried_server_list = {}
-                while true do
-                    server, info = servers_heap:peek()
-                    -- we need to let the retry > #nodes so this branch can be hit and
-                    -- the request will retry next priority of nodes
-                    if server == nil then
-                        err = "all upstream servers tried"
-                        break
-                    end
-
-                    if not ctx.balancer_tried_servers[server] then
-                        break
-                    end
-
-                    servers_heap:pop()
-                    core.table.insert(tried_server_list, info)
+            local skipped
+
+            while true do
+                server, info = servers_heap:peek()
+                -- we need to let the retry > #nodes so this branch can be hit and
+                -- the request will retry next priority of nodes
+                if server == nil then
+                    err = "all upstream servers tried"
+                    break
                 end
 
-                for _, info in ipairs(tried_server_list) do
-                    servers_heap:insert(info, info.server)
+                -- the heap is shared with the pickers built for later versions of
+                -- the upstream, so it can hold nodes this request's conf does not
+                -- know about. Only hand out the ones it does
+                if up_nodes[server] and not (tried and tried[server]) then
+                    break
+                end
+
+                servers_heap:pop()
+                if not skipped then
+                    skipped = {}
+                end
+                core.table.insert(skipped, info)
+            end
+
+            if skipped then
+                for _, skipped_info in ipairs(skipped) do
+                    servers_heap:insert(skipped_info, skipped_info.server)
                 end
-            else
-                server, info = servers_heap:peek()
             end
 
             if not server then
                 return nil, err
             end
 
-            info.score = info.score + info.effect_weight
-            servers_heap:update(server, info)
+            conns[server] = (conns[server] or 0) + 1
+            update_score(state, server)
             return server
         end,
         after_balance = function (ctx, before_retry)
+            -- the release is what makes the request stop holding the server, so drop
+            -- the reference here instead of leaving every caller to remember. A caller
+            -- that goes on to retry gets a fresh one from the next pick
             local server = ctx.balancer_server
-            local info = servers_heap:valueByPayload(server)
-            info.score = info.score - info.effect_weight
-            servers_heap:update(server, info)
+            ctx.balancer_server = nil
+
+            if server then
+                local count = (conns[server] or 0) - 1
+                if count < 0 then
+                    -- a release with no matching pick. The store below floors the
+                    -- count either way, so the score cannot be corrupted - but the
+                    -- accounting is wrong and it should not pass in silence
+                    core.log.error("released a connection never picked on ", server)
+                end
+                conns[server] = count > 0 and count or nil
+                update_score(state, server)
+            end
 
             if not before_retry then
                 if ctx.balancer_tried_servers then
@@ -94,6 +212,10 @@ function _M.new(up_nodes, upstream)
                 return nil
             end
 
+            if not server then
+                return nil
+            end
+
             if not ctx.balancer_tried_servers then
                 ctx.balancer_tried_servers = core.tablepool.fetch("balancer_tried_servers", 0, 2)
             end
diff --git a/apisix/balancer/priority.lua b/apisix/balancer/priority.lua
index af5d60cbbeb1..40e65f3bc393 100644
--- a/apisix/balancer/priority.lua
+++ b/apisix/balancer/priority.lua
@@ -33,7 +33,9 @@ function _M.new(up_nodes, upstream, picker_mod)
 
     local pickers = core.table.new(#priority_index, 0)
     for i, priority in ipairs(priority_index) do
-        local picker, err = picker_mod.new(up_nodes[priority], upstream)
+        -- the priority is part of the picker's identity: node sets of different
+        -- priorities are disjoint and must not share balancing state
+        local picker, err = picker_mod.new(up_nodes[priority], upstream, priority)
         if not picker then
             return nil, "failed to create picker with priority " .. priority .. ": " .. err
         end
diff --git a/apisix/balancer/roundrobin.lua b/apisix/balancer/roundrobin.lua
index 7090f526117c..ca97e9916b01 100644
--- a/apisix/balancer/roundrobin.lua
+++ b/apisix/balancer/roundrobin.lua
@@ -58,6 +58,11 @@ function _M.new(up_nodes, upstream)
             return server
         end,
         after_balance = function (ctx, before_retry)
+            -- the release is what makes the request stop holding the server, so drop
+            -- the reference here instead of leaving every caller to remember
+            local server = ctx.balancer_server
+            ctx.balancer_server = nil
+
             if not before_retry then
                 if ctx.balancer_tried_servers then
                     core.tablepool.release("balancer_tried_servers", ctx.balancer_tried_servers)
@@ -67,11 +72,15 @@ function _M.new(up_nodes, upstream)
                 return nil
             end
 
+            if not server then
+                return nil
+            end
+
             if not ctx.balancer_tried_servers then
                 ctx.balancer_tried_servers = core.tablepool.fetch("balancer_tried_servers", 0, 2)
             end
 
-            ctx.balancer_tried_servers[ctx.balancer_server] = true
+            ctx.balancer_tried_servers[server] = true
             ctx.balancer_tried_servers_count = (ctx.balancer_tried_servers_count or 0) + 1
         end,
         before_retry_next_priority = function (ctx)
diff --git a/apisix/init.lua b/apisix/init.lua
index 162affd3adc9..3c0ff689f55b 100644
--- a/apisix/init.lua
+++ b/apisix/init.lua
@@ -1541,6 +1541,10 @@ function _M.stream_log_phase()
 
     healthcheck_passive(api_ctx)
 
+    if api_ctx.server_picker and api_ctx.server_picker.after_balance then
+        api_ctx.server_picker.after_balance(api_ctx, false)
+    end
+
     core.ctx.release_vars(api_ctx)
     if api_ctx.plugins then
         core.tablepool.release("plugins", api_ctx.plugins)
diff --git a/apisix/upstream.lua b/apisix/upstream.lua
index ef89f2a22f84..232b0eb0dd48 100644
--- a/apisix/upstream.lua
+++ b/apisix/upstream.lua
@@ -113,6 +113,22 @@ local scheme_to_port = {
 _M.scheme_to_port = scheme_to_port
 
 
+-- A bare (unbracketed) IPv6 host makes the "host:port" key the balancer and the
+-- health checker build ambiguous (parse_addr reads the whole thing as an address
+-- with no port). The Admin API rejects such hosts and check_upstream_conf brackets
+-- them for configured upstreams, but service discovery returns nodes that reach
+-- here without going through either.
+--
+-- Only a node that carries its own port is bracketed. A host given without one is
+-- itself ambiguous - the map key "::1:1980" is a valid IPv6 literal as much as it
+-- is host ::1 port 1980 - so it is left to the existing malformed-config handling.
+-- Discovery endpoints always carry a port, so nothing real is missed.
+local function needs_ipv6_bracket(node)
+    return node.port and core.utils.parse_ipv6(node.host)
+                     and str_byte(node.host, 1) ~= str_byte("[")
+end
+
+
 local function fill_node_info(up_conf, scheme, is_stream)
     local nodes = up_conf.nodes
     if up_conf.nodes_ref == nodes then
@@ -135,6 +151,10 @@ local function fill_node_info(up_conf, scheme, is_stream)
         if not n.priority then
             need_filled = true
         end
+
+        if needs_ipv6_bracket(n) then
+            need_filled = true
+        end
     end
 
     if not need_filled then
@@ -148,10 +168,12 @@ local function fill_node_info(up_conf, scheme, is_stream)
     -- keep the original nodes for slow path in `compare_upstream_node()`,
     -- can't use `core.table.deepcopy()` for whole `nodes` array here,
     -- because `compare_upstream_node()` compare `metadata` of node by address.
+    -- The original (bare) host is preserved there, so bracketing below does not
+    -- make discovery re-fetches look like a node change.
     up_conf.original_nodes = core.table.new(#nodes, 0)
     for i, n in ipairs(nodes) do
         up_conf.original_nodes[i] = core.table.clone(n)
-        if not n.port or not n.priority then
+        if not n.port or not n.priority or needs_ipv6_bracket(n) then
             nodes[i] = core.table.clone(n)
 
             if not is_stream and not n.port then
@@ -162,6 +184,10 @@ local function fill_node_info(up_conf, scheme, is_stream)
             if not n.priority then
                 nodes[i].priority = 0
             end
+
+            if needs_ipv6_bracket(n) then
+                nodes[i].host = "[" .. n.host .. "]"
+            end
         end
     end
 
diff --git a/t/lib/server.lua b/t/lib/server.lua
index d933478cdc6f..71b34912a5d7 100644
--- a/t/lib/server.lua
+++ b/t/lib/server.lua
@@ -386,6 +386,30 @@ end
 _M.websocket_handshake_route = _M.websocket_handshake
 
 
+-- keep the session open until the peer goes away, so that the request stays in
+-- flight in the balancer the way a real WebSocket session does. An idle timeout is
+-- the normal state of such a session, not an error: keep waiting, and only give up
+-- once the peer closes or the connection breaks
+function _M.websocket_hold()
+    local websocket = require "resty.websocket.server"
+    local wb, err = websocket:new({timeout = 30000})
+    if not wb then
+        ngx.log(ngx.ERR, "failed to new websocket: ", err)
+        return ngx.exit(400)
+    end
+
+    while true do
+        local _, typ, err = wb:recv_frame()
+        if typ == "close" then
+            return
+        end
+        if not typ and not string.find(err or "", "timeout", 1, true) then
+            return
+        end
+    end
+end
+
+
 function _M.api_breaker()
     ngx.exit(tonumber(ngx.var.arg_code))
 end
diff --git a/t/node/least_conn.t b/t/node/least_conn.t
index 174252fd713d..cf79b26747e0 100644
--- a/t/node/least_conn.t
+++ b/t/node/least_conn.t
@@ -149,3 +149,40 @@ qr/proxy request to \S+ while connecting to upstream/
 --- grep_error_log_out
 proxy request to 127.0.0.1:1999 while connecting to upstream
 proxy request to 0.0.0.0:1999 while connecting to upstream
+
+
+
+=== TEST 5: more retries than nodes, the request ends up holding no server
+--- apisix_yaml
+upstreams:
+  - id: 1
+    type: least_conn
+    retries: 3
+    nodes:
+        "127.0.0.1:1999": 2
+        "0.0.0.0:1999": 1
+--- error_code: 502
+--- error_log
+failed to find valid upstream server, all upstream servers tried
+--- no_error_log
+table index is nil
+
+
+
+=== TEST 6: a single node is retried, it is the only one there is
+--- apisix_yaml
+upstreams:
+  - id: 1
+    type: least_conn
+    retries: 2
+    nodes:
+        "127.0.0.1:1999": 1
+--- error_code: 502
+--- error_log
+connect() failed
+--- grep_error_log eval
+qr/proxy request to \S+ while connecting to upstream/
+--- grep_error_log_out
+proxy request to 127.0.0.1:1999 while connecting to upstream
+proxy request to 127.0.0.1:1999 while connecting to upstream
+proxy request to 127.0.0.1:1999 while connecting to upstream
diff --git a/t/node/least_conn3.t b/t/node/least_conn3.t
new file mode 100644
index 000000000000..723108f221f2
--- /dev/null
+++ b/t/node/least_conn3.t
@@ -0,0 +1,575 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+use t::APISIX 'no_plan';
+
+repeat_each(2);
+log_level('info');
+no_root_location();
+no_shuffle();
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: keep in-flight conn count across balancer recreation on scaling
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            -- resource_key is stable across scaling, so both pickers share the
+            -- same connection count table
+            local up = {resource_key = "/upstreams/lc-scale"}
+
+            -- 2 nodes serving long-lived connections
+            local nodes = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1}
+            local p1 = least_conn.new(nodes, up, 0)
+
+            -- establish 4 in-flight connections (get without after_balance)
+            local ctx = {}
+            local held = {}
+            for _ = 1, 4 do
+                held[#held + 1] = p1.get(ctx)
+            end
+
+            -- scale out: add a third node, the picker is recreated
+            local scaled = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1,
+                            ["127.0.0.1:1982"] = 1}
+            local p2 = least_conn.new(scaled, up, 0)
+
+            -- the freshly added node has no connection, so it must be picked first
+            for _ = 1, 2 do
+                local s = p2.get(ctx)
+                held[#held + 1] = s
+                ngx.say(s)
+            end
+
+            -- release everything so repeated runs start from a clean state
+            for _, s in ipairs(held) do
+                ctx.balancer_server = s
+                p2.after_balance(ctx, false)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+127.0.0.1:1982
+127.0.0.1:1982
+
+
+
+=== TEST 2: a drained node returns to the pool across balancer recreation
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            local up = {resource_key = "/upstreams/lc-drain"}
+
+            local nodes = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1}
+            local p1 = least_conn.new(nodes, up, 0)
+
+            -- hold 4 in-flight connections, 2 land on each node
+            local ctx = {}
+            local held = {}
+            for _ = 1, 4 do
+                held[#held + 1] = p1.get(ctx)
+            end
+
+            -- drain every connection on 1980
+            for _, s in ipairs(held) do
+                if s == "127.0.0.1:1980" then
+                    ctx.balancer_server = s
+                    p1.after_balance(ctx, false)
+                end
+            end
+
+            -- picker recreated: 1980 is back to baseline and must be preferred
+            -- over 1981 which is still holding connections
+            local p2 = least_conn.new(nodes, up, 0)
+            local s = p2.get(ctx)
+            ngx.say(s)
+
+            -- release the rest so repeated runs start from a clean state
+            ctx.balancer_server = s
+            p2.after_balance(ctx, false)
+            for _, h in ipairs(held) do
+                if h == "127.0.0.1:1981" then
+                    ctx.balancer_server = h
+                    p2.after_balance(ctx, false)
+                end
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+127.0.0.1:1980
+
+
+
+=== TEST 3: scale down drops the removed node, remaining nodes balance
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            local up = {resource_key = "/upstreams/lc-scale-down"}
+
+            local nodes = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1}
+            local p1 = least_conn.new(nodes, up, 0)
+
+            local ctx = {}
+            -- fully complete two requests, one per node
+            for _ = 1, 2 do
+                local s = p1.get(ctx)
+                ctx.balancer_server = s
+                p1.after_balance(ctx, false)
+            end
+
+            -- scale down to a single remaining node, picker recreated
+            local scaled = {["127.0.0.1:1981"] = 1}
+            local p2 = least_conn.new(scaled, up, 0)
+
+            local s = p2.get(ctx)
+            ctx.balancer_server = s
+            p2.after_balance(ctx, false)
+            ngx.say(s)
+        }
+    }
+--- request
+GET /t
+--- response_body
+127.0.0.1:1981
+
+
+
+=== TEST 4: connections released on the old picker are seen by the new one
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            local up = {resource_key = "/upstreams/lc-drain-old-picker"}
+
+            local nodes = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1}
+            local p1 = least_conn.new(nodes, up, 0)
+
+            -- 4 long-lived connections, 2 on each node
+            local ctx = {}
+            local held = {}
+            for _ = 1, 4 do
+                held[#held + 1] = p1.get(ctx)
+            end
+
+            -- scale out: new requests are routed with a freshly built picker
+            local scaled = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1,
+                            ["127.0.0.1:1982"] = 1}
+            local p2 = least_conn.new(scaled, up, 0)
+
+            -- the long-lived connections close. They are still bound to the picker
+            -- they were routed with (ctx.server_picker), so they are released on p1
+            for _, s in ipairs(held) do
+                ctx.balancer_server = s
+                p1.after_balance(ctx, false)
+            end
+
+            -- every node is empty again, so the next requests must spread over all
+            -- of them instead of piling up on the node that was added last
+            local picked = {}
+            for _ = 1, 3 do
+                picked[#picked + 1] = p2.get(ctx)
+            end
+            table.sort(picked)
+            ngx.say(table.concat(picked, " "))
+
+            for _, s in ipairs(picked) do
+                ctx.balancer_server = s
+                p2.after_balance(ctx, false)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+127.0.0.1:1980 127.0.0.1:1981 127.0.0.1:1982
+
+
+
+=== TEST 5: node sets of different priorities keep their own state
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            local up = {resource_key = "/upstreams/lc-priority"}
+
+            local high = least_conn.new({["127.0.0.1:1980"] = 1}, up, 1)
+            local low = least_conn.new({["127.0.0.1:1981"] = 1}, up, 0)
+
+            local ctx = {}
+            for _, picker in ipairs({high, low}) do
+                local s = picker.get(ctx)
+                ngx.say(s)
+                ctx.balancer_server = s
+                picker.after_balance(ctx, false)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+127.0.0.1:1980
+127.0.0.1:1981
+
+
+
+=== TEST 6: a request that runs out of servers does not release twice
+--- config
+    location /t {
+        content_by_lua_block {
+            local balancer = require("apisix.balancer")
+            local least_conn = require("apisix.balancer.least_conn")
+
+            local up_conf = {
+                type = "least_conn",
+                resource_key = "/upstreams/lc-double-release",
+                nodes = {
+                    {host = "127.0.0.1", port = 1980, weight = 10, priority = 0},
+                    {host = "127.0.0.1", port = 1981, weight = 1, priority = 0},
+                },
+            }
+            local nodes = {["127.0.0.1:1980"] = 10, ["127.0.0.1:1981"] = 1}
+
+            -- three long-lived connections land on 1981 while it is the only node
+            local seeded = {}
+            local seed = least_conn.new({["127.0.0.1:1981"] = 1}, up_conf, 0)
+            for i = 1, 3 do
+                seeded[i] = {}
+                seeded[i].balancer_server = seed.get(seeded[i])
+            end
+
+            -- a request that fails on every node, driven through the real balancer
+            -- a fresh version, as a real scale out would produce, so the balancer
+            -- builds a picker for the two nodes over the state seeded above
+            local ctx = {upstream_conf = up_conf, upstream_version = tostring(ngx.now()),
+                         upstream_key = "lc-double-release", var = {}}
+            assert(balancer.pick_server(nil, ctx), "first try")
+            assert(balancer.pick_server(nil, ctx), "retry")
+            local server, err = balancer.pick_server(nil, ctx)
+            ngx.say(server == nil and err or "expected to run out of servers")
+
+            -- the request holds nothing now, so the log phase must release nothing
+            ngx.say("holds a server: ", ctx.balancer_server ~= nil)
+            ctx.server_picker.after_balance(ctx, false)
+
+            -- 1981 still holds the three connections, so 1980 (ten times the weight)
+            -- must win every pick below. It only ties once 1981 is thought to be
+            -- lighter than it is
+            local picked = {}
+            local stolen = false
+            local p = least_conn.new(nodes, up_conf, 0)
+            for i = 1, 35 do
+                picked[i] = {}
+                picked[i].balancer_server = p.get(picked[i])
+                if picked[i].balancer_server == "127.0.0.1:1981" then
+                    stolen = true
+                end
+            end
+            ngx.say(stolen and "stolen" or "kept")
+
+            for _, c in ipairs(picked) do
+                p.after_balance(c, false)
+            end
+            for _, c in ipairs(seeded) do
+                p.after_balance(c, false)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+failed to find valid upstream server, all upstream servers tried
+holds a server: false
+kept
+
+
+
+=== TEST 7: scaling out an upstream with in-flight requests prefers the new node
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local http = require "resty.http"
+
+            -- the requests have to outlast the whole test, and the default upstream
+            -- read timeout (6s) would cut them short well before that
+            local function set_upstream(nodes)
+                local code, body = t('/apisix/admin/upstreams/1', ngx.HTTP_PUT,
+                     [[{"type": "least_conn",
+                        "timeout": {"connect": 60, "send": 60, "read": 60},
+                        "nodes": ]] .. nodes .. [[}]])
+                assert(code < 300, body)
+            end
+
+            set_upstream([[{"127.0.0.1:1980": 1, "0.0.0.0:1980": 1}]])
+            local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT,
+                 [[{"uri": "/mysleep", "upstream_id": "1"}]])
+            assert(code < 300, body)
+
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/mysleep?seconds="
+            -- let the route and the upstream reach the router before hitting them
+            ngx.sleep(1)
+
+            -- four in-flight requests, two on each of the original nodes
+            local threads = {}
+            for i = 1, 4 do
+                threads[i] = assert(ngx.thread.spawn(function ()
+                    http.new():request_uri(uri .. "60")
+                end))
+            end
+            ngx.sleep(1)
+
+            -- scale out while they are still in flight
+            set_upstream([[{"127.0.0.1:1980": 1, "0.0.0.0:1980": 1, "127.0.0.2:1980": 1}]])
+            ngx.sleep(1)
+
+            -- the new node holds no connection, so it must take the next ones
+            for _ = 1, 2 do
+                assert(http.new():request_uri(uri .. "0.1"))
+            end
+
+            -- drop the clients. This only kills the client coroutines: the requests
+            -- they made stay parked on the upstream, so their counts are released
+            -- whenever those finish, not here. That is fine - a repeated run rebuilds
+            -- the upstream from one node, which drops the second one from the heap
+            -- and brings it back empty
+            for _, th in ipairs(threads) do
+                ngx.thread.kill(th)
+            end
+        }
+    }
+--- request
+GET /t
+--- timeout: 10
+--- grep_error_log eval
+qr/proxy request to \S+/
+--- grep_error_log_out eval
+qr/\A(?:proxy request to (?:127\.0\.0\.1|0\.0\.0\.0):1980\n){4}(?:proxy request to 127\.0\.0\.2:1980\n){2}\z/
+
+
+
+=== TEST 8: an upstream with no stable key keeps its state private to the picker
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            -- conf_server and ai-proxy-multi build pickers from an upstream table
+            -- that carries no resource key, so there is nothing to share state on
+            local nodes = {["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1}
+
+            local p1 = least_conn.new(nodes, {})
+            local ctx1 = {}
+            ctx1.balancer_server = p1.get(ctx1)
+
+            -- a second picker starts empty: it must not see the connection held on
+            -- the first one, so it picks the very same node
+            local p2 = least_conn.new(nodes, {})
+            local ctx2 = {}
+            ctx2.balancer_server = p2.get(ctx2)
+            ngx.say(ctx2.balancer_server == ctx1.balancer_server and "private" or "shared")
+
+            p1.after_balance(ctx1, false)
+            p2.after_balance(ctx2, false)
+        }
+    }
+--- request
+GET /t
+--- response_body
+private
+
+
+
+=== TEST 9: scaling out with live WebSocket sessions prefers the new node
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local ws_client = require("resty.websocket.client")
+
+            -- an upstream of its own: the balancing state is shared by every picker
+            -- built for one upstream, so a test that asserts on it cannot reuse the
+            -- upstream another test has been routing connections to
+            local function set_upstream(nodes)
+                local code, body = t('/apisix/admin/upstreams/2', ngx.HTTP_PUT,
+                     [[{"type": "least_conn",
+                        "timeout": {"connect": 60, "send": 60, "read": 60},
+                        "nodes": ]] .. nodes .. [[}]])
+                assert(code < 300, body)
+            end
+
+            set_upstream([[{"127.0.0.1:1980": 1, "0.0.0.0:1980": 1}]])
+            local code, body = t('/apisix/admin/routes/2', ngx.HTTP_PUT,
+                 [[{"uri": "/websocket_hold", "enable_websocket": true, "upstream_id": "2"}]])
+            assert(code < 300, body)
+
+            local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/websocket_hold"
+            ngx.sleep(1)
+
+            -- a session stays in flight until it is closed, which is what makes the
+            -- load stick to the nodes that were there before the scale out
+            local function hold()
+                local wb = ws_client:new({timeout = 60000})
+                assert(wb:connect(uri))
+                ngx.sleep(60)
+            end
+
+            local threads = {}
+            for i = 1, 4 do
+                threads[i] = assert(ngx.thread.spawn(hold))
+            end
+            ngx.sleep(1)
+
+            -- scale out while the four sessions are still connected
+            set_upstream([[{"127.0.0.1:1980": 1, "0.0.0.0:1980": 1, "127.0.0.2:1980": 1}]])
+            ngx.sleep(1)
+
+            -- the new node carries no session, so it must take the next ones
+            for i = 5, 6 do
+                threads[i] = assert(ngx.thread.spawn(hold))
+            end
+            ngx.sleep(1)
+
+            -- drop the clients. This only kills the client coroutines: the requests
+            -- they made stay parked on the upstream, so their counts are released
+            -- whenever those finish, not here. That is fine - a repeated run rebuilds
+            -- the upstream from one node, which drops the second one from the heap
+            -- and brings it back empty
+            for _, th in ipairs(threads) do
+                ngx.thread.kill(th)
+            end
+        }
+    }
+--- request
+GET /t
+--- timeout: 10
+--- grep_error_log eval
+qr/proxy request to \S+/
+--- grep_error_log_out eval
+qr/\A(?:proxy request to (?:127\.0\.0\.1|0\.0\.0\.0):1980\n){4}(?:proxy request to 127\.0\.0\.2:1980\n){2}\z/
+
+
+
+=== TEST 10: scaling a single-node upstream out counts the sessions it already has
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local http = require "resty.http"
+
+            -- the sessions have to outlast the whole test, and the default upstream
+            -- read timeout (6s) would cut them short well before that
+            local function set_upstream(nodes)
+                local code, body = t('/apisix/admin/upstreams/3', ngx.HTTP_PUT,
+                     [[{"type": "least_conn",
+                        "timeout": {"connect": 60, "send": 60, "read": 60},
+                        "nodes": ]] .. nodes .. [[}]])
+                assert(code < 300, body)
+            end
+
+            -- a single node is the state a k8s deployment or a discovery service
+            -- starts from, and the requests routed while it was alone still have to
+            -- be counted, or the scale out cannot see that it is loaded
+            set_upstream([[{"127.0.0.1:1980": 1}]])
+            -- reuse route 1: two routes cannot both own /mysleep
+            local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT,
+                 [[{"uri": "/mysleep", "upstream_id": "3"}]])
+            assert(code < 300, body)
+
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/mysleep?seconds="
+            ngx.sleep(1)
+
+            local threads = {}
+            for i = 1, 4 do
+                threads[i] = assert(ngx.thread.spawn(function ()
+                    http.new():request_uri(uri .. "60")
+                end))
+            end
+            ngx.sleep(1)
+
+            set_upstream([[{"127.0.0.1:1980": 1, "127.0.0.2:1980": 1}]])
+            ngx.sleep(1)
+
+            -- the lone node holds four requests, the new one holds none
+            for _ = 1, 4 do
+                assert(http.new():request_uri(uri .. "0.1"))
+            end
+
+            -- abort the sessions rather than outwait them: the log phase still runs,
+            -- so the counts are released before the next run of this block
+            for _, th in ipairs(threads) do
+                ngx.thread.kill(th)
+            end
+        }
+    }
+--- request
+GET /t
+--- timeout: 10
+--- grep_error_log eval
+qr/proxy request to \S+/
+--- grep_error_log_out eval
+qr/\A(?:proxy request to 127\.0\.0\.1:1980\n){4}(?:proxy request to 127\.0\.0\.2:1980\n){4}\z/
+
+
+
+=== TEST 11: a weight change is applied without losing the connection count
+--- config
+    location /t {
+        content_by_lua_block {
+            local least_conn = require("apisix.balancer.least_conn")
+            local up = {resource_key = "/upstreams/lc-reweight"}
+
+            local p1 = least_conn.new({["127.0.0.1:1980"] = 1, ["127.0.0.1:1981"] = 1}, up, 0)
+
+            -- one connection on each node: equal weight, equal score
+            local held = {}
+            for i = 1, 2 do
+                held[i] = {}
+                held[i].balancer_server = p1.get(held[i])
+            end
+
+            -- 1980 is given ten times the weight. It carries the same load as 1981,
+            -- so it is now the lighter of the two and must win the next picks
+            local p2 = least_conn.new({["127.0.0.1:1980"] = 10, ["127.0.0.1:1981"] = 1}, up, 0)
+
+            local picked = {}
+            for i = 1, 3 do
+                picked[i] = {}
+                picked[i].balancer_server = p2.get(picked[i])
+                ngx.say(picked[i].balancer_server)
+            end
+
+            for _, c in ipairs(picked) do
+                p2.after_balance(c, false)
+            end
+            for _, c in ipairs(held) do
+                p2.after_balance(c, false)
+            end
+        }
+    }
+--- request
+GET /t
+--- response_body
+127.0.0.1:1980
+127.0.0.1:1980
+127.0.0.1:1980
diff --git a/t/node/upstream-discovery-ipv6.t b/t/node/upstream-discovery-ipv6.t
new file mode 100644
index 000000000000..438615a7a17e
--- /dev/null
+++ b/t/node/upstream-discovery-ipv6.t
@@ -0,0 +1,192 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+use t::APISIX 'no_plan';
+
+repeat_each(1);
+log_level('info');
+no_root_location();
+no_shuffle();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!$block->yaml_config) {
+        $block->set_value("yaml_config", <<_EOC_);
+apisix:
+    node_listen: 1984
+deployment:
+    role: data_plane
+    role_data_plane:
+        config_provider: yaml
+_EOC_
+    }
+
+    # a discovery node reaches the balancer without going through the schema check
+    # or the bracketing the Admin API applies, so a bare IPv6 host arrives here raw
+    $block->set_value("listen_ipv6", 1);
+
+    if ($block->apisix_yaml) {
+        $block->set_value("apisix_yaml", $block->apisix_yaml . <<_EOC_);
+upstreams:
+    - service_name: mock
+      discovery_type: mock
+      type: least_conn
+      id: 1
+#END
+_EOC_
+    }
+
+    if (!$block->request) {
+        $block->set_value("request", "GET /t");
+    }
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: single bare-IPv6 node from discovery, least_conn
+--- apisix_yaml
+routes:
+  - uris:
+      - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            local discovery = require("apisix.discovery.init").discovery
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "::1", port = 1980, weight = 1},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local res = assert(http.new():request_uri(uri, {keepalive = false}))
+            ngx.say(res.status)
+            ngx.print(res.body)
+        }
+    }
+--- response_body
+200
+hello world
+--- no_error_log
+attempt to concatenate
+
+
+
+=== TEST 2: multiple bare-IPv6 nodes from discovery, least_conn
+--- apisix_yaml
+routes:
+  - uris:
+      - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            local discovery = require("apisix.discovery.init").discovery
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "::1", port = 1980, weight = 1},
+                        {host = "0:0:0:0:0:0:0:1", port = 1980, weight = 1},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local res = assert(http.new():request_uri(uri, {keepalive = false}))
+            ngx.say(res.status)
+            ngx.print(res.body)
+        }
+    }
+--- response_body
+200
+hello world
+--- no_error_log
+attempt to concatenate
+
+
+
+=== TEST 3: a discovery re-fetch of the same bare-IPv6 nodes is not seen as a change
+--- apisix_yaml
+routes:
+  - uris:
+      - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            local discovery = require("apisix.discovery.init").discovery
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "::1", port = 1980, weight = 1},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            for _ = 1, 3 do
+                assert(http.new():request_uri(uri, {keepalive = false}))
+            end
+            ngx.say("ok")
+        }
+    }
+--- response_body
+ok
+--- grep_error_log eval
+qr/create_obj_fun\(\): upstream nodes:/
+--- grep_error_log_out
+create_obj_fun(): upstream nodes:
+
+
+
+=== TEST 4: bare-IPv6 node that already has port and priority is still bracketed
+--- apisix_yaml
+routes:
+  - uris:
+      - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            -- port and priority both set, so the node skips the port/priority fill.
+            -- Only the IPv6 branch of the clone condition can trigger here
+            local discovery = require("apisix.discovery.init").discovery
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "::1", port = 1980, weight = 1, priority = 0},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local res = assert(http.new():request_uri(uri, {keepalive = false}))
+            ngx.say(res.status)
+            ngx.print(res.body)
+        }
+    }
+--- response_body
+200
+hello world
+--- no_error_log
+attempt to concatenate
diff --git a/t/stream-node/least_conn.t b/t/stream-node/least_conn.t
new file mode 100644
index 000000000000..15f8747be572
--- /dev/null
+++ b/t/stream-node/least_conn.t
@@ -0,0 +1,74 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+use t::APISIX 'no_plan';
+
+# every repeat opens a new connection against the same worker, so the balancing
+# state built by the previous ones is still there. Without releasing the server
+# in the log phase the counts only grow and the picks start to drift away from
+# the node with the highest weight.
+repeat_each(4);
+log_level('info');
+no_root_location();
+no_shuffle();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if ($block->apisix_yaml && !$block->yaml_config) {
+        my $yaml_config = <<_EOC_;
+apisix:
+    node_listen: 1984
+deployment:
+    role: data_plane
+    role_data_plane:
+        config_provider: yaml
+_EOC_
+
+        $block->set_value("yaml_config", $yaml_config);
+    }
+
+    $block->set_value("stream_enable", 1);
+
+    if (!$block->stream_request) {
+        $block->set_value("stream_request", "mmm");
+    }
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: release the finished connection in the stream log phase
+--- apisix_yaml
+stream_routes:
+  - id: 1
+    upstream:
+        type: least_conn
+        nodes:
+        - host: 127.0.0.1
+          port: 1995
+          weight: 3
+        - host: 127.0.0.2
+          port: 1995
+          weight: 1
+#END
+--- stream_response
+hello world
+--- grep_error_log eval
+qr/proxy request to \S+/
+--- grep_error_log_out
+proxy request to 127.0.0.1:1995