test(dst): pin remaining decision randomness and serialize bubble scheduling

.github/workflows/ci.yml

@@ -42,6 +42,11 @@ jobs:
       # go test surfaces captured stderr only on failure, so per-node debug
       # logs are free on pass and present when a rare flake finally trips.
       DST_LOG: '1'
+      # Serialize scheduling inside the synctest bubble: one P runs goroutines
+      # in runqueue order and async preemption stays off, so same-tick wakeups
+      # replay in program order instead of racing across threads.
+      GOMAXPROCS: '1'
+      GODEBUG: 'asyncpreemptoff=1'
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5

.github/workflows/live-i2p-roundtrip.yml

@@ -28,4 +28,4 @@ jobs:
           set -euo pipefail
           IVNP_LIVE_ROUNDTRIP=1 \
             go test -tags integration -v -timeout=18m \
-              -run '^TestLiveI2PRoundTrip$' .
+              -run '^TestLiveI2PRoundTrip$' ./tests/integration

client/internal/addressbook/subscription.go

@@ -9,8 +9,8 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
-	"sync"
 
+	"gosuda.org/ivnp/internal/durable"
 	"gosuda.org/ivnp/internal/parallelism"
 )
 
@@ -77,7 +77,7 @@ func (s *Service) refresh(parent context.Context) error {
 		fetches[index] = fetchJob{raw: raw, haveSnapshot: haveSnapshot, etag: etags[raw], modified: modified[raw]}
 	}
 	workers := parallelism.Workers(len(fetches))
-	var group sync.WaitGroup
+	var group durable.WaitGroup
 	group.Add(workers)
 	for range workers {
 		go func() {

client/internal/frontend/server.go

@@ -7,6 +7,8 @@ import (
 	"net"
 	"net/http"
 	"sync"
+
+	"gosuda.org/ivnp/internal/durable"
 )
 
 var (
@@ -20,7 +22,7 @@ type server struct {
 	context    context.Context
 	cancel     context.CancelFunc
 	done       chan struct{}
-	activities sync.WaitGroup
+	activities durable.WaitGroup
 	started    bool
 	closed     bool
 }

client/internal/sam/server.go

@@ -75,7 +75,7 @@ type Server struct {
 	ctx          context.Context
 	cancel       context.CancelFunc
 	done         chan struct{}
-	wg           sync.WaitGroup
+	wg           durable.WaitGroup
 	started      bool
 	closed       bool
 	sem          chan struct{}

client/internal/sam/session.go

@@ -11,6 +11,7 @@ import (
 	"gosuda.org/ivnp/dataplane"
 	"gosuda.org/ivnp/foundation"
 	"gosuda.org/ivnp/interfaces/destination"
+	"gosuda.org/ivnp/internal/durable"
 )
 
 var errUnknownSubsession = errors.New("sam: unknown subsession")
@@ -76,7 +77,7 @@ type samSession struct {
 	acceptCancellations atomic.Uint64
 	once                sync.Once
 	closeErr            error
-	wg                  sync.WaitGroup
+	wg                  durable.WaitGroup
 }
 
 func newRootSession(server *Server, id string, style sessionStyle, endpoint destination.DestinationEndpoint, control *serverConnection, fromPort, toPort, listenPort uint16, protocol, listenProtocol uint8, rawHeader bool, udpTarget *net.UDPAddr, offline *foundation.OfflineSignature) *samSession {

controlplane/dst_seed.go

@@ -10,10 +10,12 @@ import (
 )
 
 // SetDeterministicSeeds pins the tunnel peer-selection keys, the netdb
-// explorer's noise stream, and the transport mux's preference draw for
-// deterministic simulation. Nil restores crypto randomness for all.
-func SetDeterministicSeeds(tunnelSeed, explorerSeed, muxSeed *foundation.Hash) {
+// explorer's noise stream, the transport mux's preference draw, and the
+// build-message deadline fuzz for deterministic simulation. Nil restores
+// crypto randomness for all.
+func SetDeterministicSeeds(tunnelSeed, explorerSeed, muxSeed, buildSeed *foundation.Hash) {
 	tunnel.SetDeterministicSelectionSeed(tunnelSeed)
 	netdb.SetDeterministicExplorerSeed(explorerSeed)
 	router.SetDeterministicMuxSeed(muxSeed)
+	tunnel.SetDeterministicBuildSeed(buildSeed)
 }

controlplane/internal/netdb/confirmation.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"gosuda.org/ivnp/foundation"
+	"gosuda.org/ivnp/internal/durable"
 )
 
 const (
@@ -310,7 +311,7 @@ func (p *confirmedPublication) maintain(ctx context.Context, force bool) (int, e
 			break
 		}
 		results := make([]error, len(batch))
-		var group sync.WaitGroup
+		var group durable.WaitGroup
 		group.Add(len(batch))
 		for index := range batch {
 			go func() {

controlplane/internal/netdb/lookup_responder.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"gosuda.org/ivnp/foundation"
+	"gosuda.org/ivnp/internal/durable"
 	"gosuda.org/ivnp/internal/parallelism"
 )
 
@@ -75,7 +76,7 @@ type LookupResponder struct {
 	started bool
 	closed  bool
 	cancel  context.CancelFunc
-	wg      sync.WaitGroup
+	wg      durable.WaitGroup
 	err     error
 }
 

controlplane/internal/netdb/publication.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"gosuda.org/ivnp/foundation"
+	"gosuda.org/ivnp/internal/durable"
 	"gosuda.org/ivnp/internal/parallelism"
 )
 
@@ -396,7 +397,7 @@ func (p *LeaseSetPublisher) publish(ctx context.Context, force bool) (int, error
 	results := make([]error, len(targets))
 	jobs := make(chan int)
 	workers := parallelism.Workers(len(targets))
-	var group sync.WaitGroup
+	var group durable.WaitGroup
 	group.Add(workers)
 	for range workers {
 		go func() {

controlplane/internal/netdb/requests.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"gosuda.org/ivnp/foundation"
+	"gosuda.org/ivnp/internal/durable"
 	"gosuda.org/ivnp/internal/parallelism"
 	"gosuda.org/ivnp/observability"
 )
@@ -151,8 +152,8 @@ type RequestManager struct {
 	mu      sync.Mutex
 	pending map[requestKey]*pendingRequest
 	closed  bool
-	active  sync.WaitGroup
-	workers sync.WaitGroup
+	active  durable.WaitGroup
+	workers durable.WaitGroup
 	jobs    chan sendWork
 	ctx     context.Context
 	cancel  context.CancelFunc

controlplane/internal/netdb/store_flooder.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"gosuda.org/ivnp/foundation"
+	"gosuda.org/ivnp/internal/durable"
 )
 
 var (
@@ -79,7 +80,7 @@ type StoreFlooder struct {
 	start     bool
 	closed    bool
 	cancel    context.CancelFunc
-	wg        sync.WaitGroup
+	wg        durable.WaitGroup
 	recent    map[[32]byte]uint64
 	keyFloods map[foundation.Hash]storeFloodCount
 }
@@ -192,7 +193,7 @@ func (f *StoreFlooder) flood(ctx context.Context, job *storeFloodJob) {
 		}
 	}
 	errs := make([]error, len(targets))
-	var sends sync.WaitGroup
+	var sends durable.WaitGroup
 	for index, target := range targets {
 		sends.Go(func() {
 			sendCtx, cancel := context.WithTimeout(ctx, storeFlooderSendTimeout)

controlplane/internal/router/route_control.go

@@ -71,7 +71,7 @@ type StreamingTunnelSender struct {
 	awaitControl         func(context.Context) error
 	tunnels              *dataplane.TunnelRuntime
 	replySlots           chan struct{}
-	replies              sync.WaitGroup
+	replies              durable.WaitGroup
 	replyGates           map[foundation.Hash]*ratchetReplyGate
 	replyGateCapacity    int
 	logger               *slog.Logger
@@ -92,7 +92,7 @@ type StreamingTunnelSender struct {
 	preparationTimeout   time.Duration
 	preparationCtx       context.Context
 	cancelPreparation    context.CancelFunc
-	preparing            sync.WaitGroup
+	preparing            durable.WaitGroup
 	seedMu               sync.Mutex
 	seedCache            [streamingSeedCacheCapacity]streamingSeedCacheEntry
 	seedNext             uint8
@@ -311,6 +311,25 @@ func (s *StreamingTunnelSender) recordFailedLeaseLocked(receipt dataplane.Router
 	s.failedLeases[path] = evidence
 }
 
+// clearFailureMarks drops recorded send and silence evidence for remote after
+// a completed refresh. An identical LeaseSet means the marks measured
+// transient loss rather than a dead remote path; a rotated LeaseSet cannot
+// match their (gateway, tunnelID) keys anyway.
+func (s *StreamingTunnelSender) clearFailureMarks(remote foundation.Hash) {
+	s.remoteMu.Lock()
+	defer s.remoteMu.Unlock()
+	for path := range s.failedRoutes {
+		if path.remote == remote {
+			delete(s.failedRoutes, path)
+		}
+	}
+	for path := range s.failedLeases {
+		if path.remote == remote {
+			delete(s.failedLeases, path)
+		}
+	}
+}
+
 func NewStreamingTunnelSender(config StreamingTunnelSenderConfig) (*StreamingTunnelSender, error) {
 	missingResolution := config.Database == nil || config.Requests == nil
 	missingTunnels := config.Tunnels == nil || config.Pool == nil
@@ -652,6 +671,7 @@ func (s *StreamingTunnelSender) resolveRoute(ctx context.Context, remote foundat
 	var lease foundation.NetworkDatabaseLease
 	var outbound controlplanetunnel.Entry
 	var circuit dataplane.TunnelCircuitInfo
+	var entries []controlplanetunnel.Entry
 	found, exhausted := false, false
 	for pass := 0; ; pass++ {
 		pick := s.leaseNext.Add(1) - 1
@@ -677,7 +697,7 @@ func (s *StreamingTunnelSender) resolveRoute(ctx context.Context, remote foundat
 		if err != nil {
 			return route, err
 		}
-		entries := s.pool.SelectableOutbound(now)
+		entries = s.pool.SelectableOutbound(now)
 		found, exhausted = false, false
 		for attempt := 0; attempt < leaseCount && !found; attempt++ {
 			if attempt != 0 {
@@ -740,8 +760,31 @@ func (s *StreamingTunnelSender) resolveRoute(ctx context.Context, remote foundat
 		} else {
 			break
 		}
+		// The fetched copy is the current truth even when byte-identical to
+		// the cached one: retrying its leases is how recovery is observed.
+		s.clearFailureMarks(remote)
 	}
 	if !found {
+		if s.logger != nil {
+			matched, marked := 0, 0
+			for _, entry := range entries {
+				if entry.Direction != controlplanetunnel.Outbound || entry.Owner != s.owner {
+					continue
+				}
+				matched++
+				candidate, ok := s.tunnels.InspectCircuit(entry.ID)
+				if !ok || candidate.Token != entry.Circuit || candidate.Owner != s.owner {
+					continue
+				}
+				s.remoteMu.RLock()
+				failedUntil := s.failedRoutes[failedRoutePath{remote: remote, circuit: entry.Circuit, gateway: lease.Gateway, tunnelID: lease.TunnelID}]
+				s.remoteMu.RUnlock()
+				if failedUntil > now {
+					marked++
+				}
+			}
+			s.logger.Debug("route resolve exhausted: circuit not found", "remote", foundation.EncodeI2PBase64(remote[:]), "entries", len(entries), "owner_matched", matched, "failure_marked", marked, "exhausted", exhausted)
+		}
 		return route, dataplane.TunnelErrCircuitNotFound
 	}
 	if route.Legacy {
@@ -1035,8 +1078,10 @@ func (s *StreamingTunnelSender) refreshRemoteLeaseSet(ctx context.Context, remot
 		return false
 	}
 	select {
-	case _, ok := <-result:
-		return ok
+	case outcome, ok := <-result:
+		// A completed lookup may still carry a terminal error; only fresh
+		// data justifies retrying marked combinations.
+		return ok && outcome.Err == nil
 	case <-ctx.Done():
 		return false
 	}