From c7baec25e263c273b42522a21c82bee3f12abab2 Mon Sep 17 00:00:00 2001 From: ehsan shariati Date: Mon, 5 Oct 2026 01:19:10 -0400 Subject: [PATCH 1/6] fix(blockchain): make pool leave clear the Blox's pool config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HandlePoolLeave always answered 500 and never cleaned up: it gated the cleanup on an eth_call of removeMemberPeerId with a placeholder selector (0x12345678) that reverts on both chains, using the caller's peer id. The device can't remove itself on-chain anyway — the app sends removeMemberPeerId from the member wallet. - HandlePoolLeave now checks the chain (strictly: RPC errors are errors) that the device's cluster peer is neither a member of nor waiting to join the pool, re-checking briefly for RPC lag, then clears pool and chain from the config and answers 202. 409 if still listed or the device is configured for another pool; 503 if the chain can't be read. Config is never cleared on a guess. - ReconcilePoolConfig (run at startup and on the 6 h loop) clears a configured pool the chain says the Blox has left — e.g. it left from the app while offline — except within 24 h of a join from this device (join-time marker), since joins write the config before the chain lists the Blox. - After the config is cleared, signal the existing fula restart marker so ipfs-cluster re-runs its init and stops following the old pool. - Real selectors for removeMemberPeerId and joinRequests; tests for the handler, reconcile and ABI helpers, plus build-tagged (livechain) read-only tests against the real SKALE and Base contracts. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QaKz9HeXaNBjrchbbGimFx --- blockchain/abi/contract.go | 36 +++- blockchain/abi/join_requests_test.go | 41 ++++ blockchain/bl_pool.go | 238 ++++++++++++++++----- blockchain/options.go | 10 + blockchain/pool_leave_live_test.go | 151 ++++++++++++++ blockchain/pool_leave_test.go | 302 +++++++++++++++++++++++++++ blox/blox.go | 12 ++ blox/pool_reconcile.go | 25 +++ 8 files changed, 764 insertions(+), 51 deletions(-) create mode 100644 blockchain/abi/join_requests_test.go create mode 100644 blockchain/pool_leave_live_test.go create mode 100644 blockchain/pool_leave_test.go create mode 100644 blox/pool_reconcile.go diff --git a/blockchain/abi/contract.go b/blockchain/abi/contract.go index 5a4b17f..bbd4d34 100644 --- a/blockchain/abi/contract.go +++ b/blockchain/abi/contract.go @@ -83,12 +83,14 @@ var MethodSignatures = struct { IsPeerIdMemberOfPool string GetMemberPeerIds string RemoveMemberPeerId string + JoinRequests string }{ Pools: "0xced08b2d", // pools(uint32) - updated signature PoolIds: "0x69883b4e", // poolIds(uint256) - index-based pool discovery IsPeerIdMemberOfPool: "0xb098a605", // isPeerIdMemberOfPool(uint32,bytes32) - corrected from block explorer GetMemberPeerIds: "0x31db3ae8", // getMemberPeerIds(uint32,address) - RemoveMemberPeerId: "0x12345678", // removeMemberPeerId(uint32,bytes32) - TODO: Update with actual signature + RemoveMemberPeerId: "0x3d71233a", // removeMemberPeerId(uint32,bytes32) — keccak256 of the signature + JoinRequests: "0xccc9fc03", // joinRequests(uint32,bytes32) — keccak256 of the signature } // DecodePoolsResult decodes the result from pools(uint32) contract call @@ -394,3 +396,35 @@ func EncodeRemoveMemberPeerIdCall(poolID uint32, peerIDBytes32 string) string { return fmt.Sprintf("%s%064x%s", MethodSignatures.RemoveMemberPeerId, poolID, peerIDBytes32) } + +// EncodeJoinRequestsCall encodes the joinRequests(uint32,bytes32) public-mapping getter call +func EncodeJoinRequestsCall(poolID uint32, peerIDBytes32 string) string { + // Remove 0x prefix from peerIDBytes32 if present + if strings.HasPrefix(peerIDBytes32, "0x") { + peerIDBytes32 = peerIDBytes32[2:] + } + + return fmt.Sprintf("%s%064x%s", MethodSignatures.JoinRequests, poolID, peerIDBytes32) +} + +// DecodeJoinRequestStatus returns the status field of a joinRequests(uint32,bytes32) result: the getter returns +// (account, poolId, timestamp, index, approvals, rejections, status, peerId), 8 words; status 1 = pending, +// 0 = no request. +func DecodeJoinRequestStatus(data string) (uint8, error) { + // Remove 0x prefix + if strings.HasPrefix(data, "0x") { + data = data[2:] + } + + // 8 * 32 bytes = 512 hex chars + if len(data) < 512 { + return 0, fmt.Errorf("insufficient data length: %d, expected 512", len(data)) + } + + status, err := strconv.ParseUint(data[6*64:7*64], 16, 8) + if err != nil { + return 0, fmt.Errorf("failed to parse join request status: %w", err) + } + + return uint8(status), nil +} diff --git a/blockchain/abi/join_requests_test.go b/blockchain/abi/join_requests_test.go new file mode 100644 index 0000000..fbd8723 --- /dev/null +++ b/blockchain/abi/join_requests_test.go @@ -0,0 +1,41 @@ +package abi + +import ( + "fmt" + "strings" + "testing" +) + +func TestEncodeJoinRequestsCall(t *testing.T) { + peer := "0x" + strings.Repeat("ab", 32) + got := EncodeJoinRequestsCall(7, peer) + want := "0xccc9fc03" + fmt.Sprintf("%064x", 7) + strings.Repeat("ab", 32) + if got != want { + t.Fatalf("EncodeJoinRequestsCall = %s, want %s", got, want) + } +} + +func TestDecodeJoinRequestStatus(t *testing.T) { + word := func(v int) string { return fmt.Sprintf("%064x", v) } + // (account, poolId, timestamp, index, approvals, rejections, status, peerId) + pending := "0x" + word(0xabc) + word(1) + word(1700000000) + word(0) + word(2) + word(0) + word(1) + strings.Repeat("cd", 32) + status, err := DecodeJoinRequestStatus(pending) + if err != nil || status != 1 { + t.Fatalf("pending request: status=%d err=%v, want 1", status, err) + } + none := "0x" + strings.Repeat(word(0), 8) + status, err = DecodeJoinRequestStatus(none) + if err != nil || status != 0 { + t.Fatalf("no request: status=%d err=%v, want 0", status, err) + } + if _, err := DecodeJoinRequestStatus("0x" + word(1)); err == nil { + t.Fatal("short data must be an error") + } +} + +func TestRemoveMemberPeerIdSelector(t *testing.T) { + // keccak256("removeMemberPeerId(uint32,bytes32)")[:4] — the old placeholder 0x12345678 matched nothing on-chain. + if MethodSignatures.RemoveMemberPeerId != "0x3d71233a" { + t.Fatalf("RemoveMemberPeerId selector = %s", MethodSignatures.RemoveMemberPeerId) + } +} diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index 1393c54..0d5b939 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -7,6 +7,8 @@ import ( "fmt" "io" "net/http" + "os" + "path/filepath" "strconv" "strings" "time" @@ -129,6 +131,8 @@ func (bl *FxBlockchain) HandlePoolJoin(method string, action string, from peer.I return } + recordPoolJoinTime(time.Now()) + statusCode := http.StatusAccepted res = PoolJoinResponse{ Account: "", @@ -324,6 +328,9 @@ func (bl *FxBlockchain) cleanLeaveJoinPool(ctx context.Context, PoolID int) { log.Errorw("Failed to reset chain name", "error", err) } } + bl.topicName = "0" + bl.chainName = "" + clearPoolJoinTime() bl.StopPingServer(ctx) // Announcements no longer used - pool joins handled via blockchain API @@ -1180,7 +1187,15 @@ func (bl *FxBlockchain) validatePoolOnChain(ctx context.Context, poolID uint32, return nil } -// HandlePoolLeave handles pool leave requests with chain support +// HandlePoolLeave handles pool leave requests with chain support. +// +// The on-chain removal (removeMemberPeerId) is sent by the app from the member's wallet — the device has no key to +// sign it — so this handler only drops the pool from the local config, and only once the chain confirms the device's +// cluster peer is neither a member of the pool nor waiting to join it: +// - 202: removed on-chain (or never there) → pool and chain cleared from config, fula services restarted so +// ipfs-cluster leaves the old pool; +// - 409: the device is configured for a different pool, or the chain still lists it (the app's tx not mined yet); +// - 503: the chain could not be read, or the cluster peer id is unknown — config untouched, safe to retry. func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer.ID, w http.ResponseWriter, r *http.Request) { log := log.With("action", action, "from", from) var req PoolLeaveRequest @@ -1197,9 +1212,25 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. ctx, cancel := context.WithTimeout(r.Context(), time.Second*time.Duration(bl.timeout)) defer cancel() + writeError := func(status int, message, description string) { + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]interface{}{"message": message, "description": description}) + } + poolID := req.PoolID chainName := req.ChainName + // Never clear a different pool than the one asked about. + configuredPool := "0" + if bl.getPoolName != nil { + configuredPool = bl.getPoolName() + } + if configuredPool != "" && configuredPool != "0" && configuredPool != strconv.Itoa(poolID) { + writeError(http.StatusConflict, "Pool mismatch", + fmt.Sprintf("This Blox is configured for pool %s, not pool %d", configuredPool, poolID)) + return + } + // If no chain name provided, try to determine it from current configuration if chainName == "" { var currentChain string @@ -1223,25 +1254,30 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. } } - // Validate that the pool exists on the specified chain - if err := bl.validatePoolOnChain(ctx, uint32(poolID), chainName); err != nil { - errMsg := map[string]interface{}{ - "message": "Pool validation failed", - "description": fmt.Sprintf("Pool %d does not exist on chain %s: %s", poolID, chainName, err.Error()), + // The app sends this right after its removeMemberPeerId tx is mined; a public RPC can lag a block or two. + stillListed := false + for attempt := 1; attempt <= poolLeaveCheckAttempts; attempt++ { + member, pending, err := bl.ClusterPeerPoolStatus(ctx, uint32(poolID), chainName) + if err != nil { + log.Warnw("Pool leave: membership check failed", "poolID", poolID, "chain", chainName, "error", err) + writeError(http.StatusServiceUnavailable, "Could not verify pool membership", + fmt.Sprintf("Membership check on %s failed: %s", chainName, err.Error())) + return } - w.WriteHeader(http.StatusBadRequest) - json.NewEncoder(w).Encode(errMsg) - return - } - - // Call the removeMemberPeerId contract method - if err := bl.callRemoveMemberPeerId(ctx, uint32(poolID), from.String(), chainName); err != nil { - errMsg := map[string]interface{}{ - "message": "Failed to remove member from pool", - "description": fmt.Sprintf("Contract call failed: %s", err.Error()), + stillListed = member || pending + if !stillListed || attempt == poolLeaveCheckAttempts { + break } - w.WriteHeader(http.StatusInternalServerError) - json.NewEncoder(w).Encode(errMsg) + select { + case <-ctx.Done(): + writeError(http.StatusServiceUnavailable, "Could not verify pool membership", ctx.Err().Error()) + return + case <-time.After(poolLeaveCheckInterval): + } + } + if stillListed { + writeError(http.StatusConflict, "Still in the pool on-chain", + fmt.Sprintf("This Blox is still a member of (or waiting to join) pool %d on %s; remove it on-chain first", poolID, chainName)) return } @@ -1261,32 +1297,114 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. if err := json.NewEncoder(w).Encode(res); err != nil { log.Error("failed to write response: %v", err) } + // Give the response time to reach the app before the services restart. + bl.signalPoolConfigCleared(poolConfigClearedRestartDelay) } -// callRemoveMemberPeerId calls the removeMemberPeerId contract method -func (bl *FxBlockchain) callRemoveMemberPeerId(ctx context.Context, poolID uint32, peerID string, chainName string) error { - chainConfigs := GetChainConfigs() - chainConfig, exists := chainConfigs[chainName] - if !exists { - return fmt.Errorf("unsupported chain: %s", chainName) +const ( + poolLeaveCheckAttempts = 3 + // PoolJoinGracePeriod is how long after a join the configured pool is kept even though the chain doesn't list + // the Blox yet: a join writes the config immediately, but the join server's addMember (or approval votes for a + // contract join) lands later. + PoolJoinGracePeriod = 24 * time.Hour + joinRequestPending = 1 // JoinRequest.status of a request still waiting for votes +) + +// Variables (not constants) so tests can shorten them. +var ( + poolLeaveCheckInterval = 4 * time.Second + poolConfigClearedRestartDelay = 3 * time.Second +) + +// PoolJoinedAtFilePath records when HandlePoolJoin last wrote a pool into the config (see PoolJoinGracePeriod). +// A variable so tests can point it at a temp dir. +var PoolJoinedAtFilePath = "/internal/.tmp/pool_joined_at.tmp" + +func recordPoolJoinTime(t time.Time) { + if err := os.MkdirAll(filepath.Dir(PoolJoinedAtFilePath), 0755); err != nil { + log.Warnw("Failed to create directory for pool join time", "err", err) + return + } + if err := os.WriteFile(PoolJoinedAtFilePath, []byte(t.UTC().Format(time.RFC3339)), 0644); err != nil { + log.Warnw("Failed to record pool join time", "err", err) + } +} + +func clearPoolJoinTime() { + if err := os.Remove(PoolJoinedAtFilePath); err != nil && !os.IsNotExist(err) { + log.Warnw("Failed to remove pool join time", "err", err) + } +} + +// PoolJoinedAt returns when the configured pool was joined from this device; false when unknown (e.g. joins made +// before this marker existed), which counts as an old join. +func PoolJoinedAt() (time.Time, bool) { + b, err := os.ReadFile(PoolJoinedAtFilePath) + if err != nil { + return time.Time{}, false + } + t, err := time.Parse(time.RFC3339, strings.TrimSpace(string(b))) + if err != nil { + return time.Time{}, false + } + return t, true +} + +// signalPoolConfigCleared runs the WithOnPoolConfigCleared hook (the blox wires it to a fula services restart, so +// ipfs-cluster re-reads the now empty pool) after delay, off the caller's goroutine. +func (bl *FxBlockchain) signalPoolConfigCleared(delay time.Duration) { + if bl.onPoolConfigCleared == nil { + return } + hook := bl.onPoolConfigCleared + go func() { + time.Sleep(delay) + hook() + }() +} - // Convert peer ID to bytes32 format - peerIDPeer, err := peer.Decode(peerID) +// ClusterPeerPoolStatus reports whether this device's cluster peer is a member of poolID on chainName, and whether +// it has a pending join request there. Unlike HandleIsMemberOfPool (which treats RPC failures as "not a member"), +// any RPC or decoding failure is returned as an error, so callers never clear the config on a guess. +func (bl *FxBlockchain) ClusterPeerPoolStatus(ctx context.Context, poolID uint32, chainName string) (member bool, pending bool, err error) { + if bl.clusterPeerID == "" { + return false, false, fmt.Errorf("cluster peer ID is not set") + } + peerIDBytes32, err := peerIdToBytes32(bl.clusterPeerID.String()) if err != nil { - return fmt.Errorf("failed to decode peer ID: %w", err) + return false, false, fmt.Errorf("failed to convert cluster peer ID to bytes32: %w", err) } - peerIDBytes32, err := PeerIDToBytes32(peerIDPeer) + memberResult, err := bl.strictEthCall(ctx, chainName, abi.EncodeIsPeerIdMemberOfPoolCall(poolID, peerIDBytes32)) + if err != nil { + return false, false, fmt.Errorf("isPeerIdMemberOfPool: %w", err) + } + membership, err := abi.DecodeIsMemberOfPoolResult(memberResult) if err != nil { - return fmt.Errorf("failed to convert peer ID to bytes32: %w", err) + return false, false, fmt.Errorf("isPeerIdMemberOfPool: %w", err) + } + if membership.IsMember { + return true, false, nil } - // Encode the removeMemberPeerId contract call - callData := abi.EncodeRemoveMemberPeerIdCall(poolID, peerIDBytes32) + requestResult, err := bl.strictEthCall(ctx, chainName, abi.EncodeJoinRequestsCall(poolID, peerIDBytes32)) + if err != nil { + return false, false, fmt.Errorf("joinRequests: %w", err) + } + status, err := abi.DecodeJoinRequestStatus(requestResult) + if err != nil { + return false, false, fmt.Errorf("joinRequests: %w", err) + } + return false, status == joinRequestPending, nil +} - // For now, we'll use eth_call to validate the call would succeed - // In a full implementation, you'd use eth_sendTransaction with proper signing +// strictEthCall runs a read-only eth_call against the pool contract on chainName and returns the raw result; RPC, +// HTTP and JSON-RPC errors are all returned as errors. +func (bl *FxBlockchain) strictEthCall(ctx context.Context, chainName string, callData string) (string, error) { + chainConfig, exists := GetChainConfigs()[chainName] + if !exists { + return "", fmt.Errorf("unsupported chain: %s", chainName) + } params := []interface{}{ map[string]interface{}{ "to": chainConfig.Contract, @@ -1294,17 +1412,13 @@ func (bl *FxBlockchain) callRemoveMemberPeerId(ctx context.Context, poolID uint3 }, "latest", } - response, statusCode, err := bl.callEVMChainWithRetry(ctx, chainName, "eth_call", params, 3) if err != nil { - return fmt.Errorf("failed to call removeMemberPeerId on chain %s: %w", chainName, err) + return "", err } - - if statusCode != 200 { - return fmt.Errorf("unexpected status code %d from chain %s", statusCode, chainName) + if statusCode != http.StatusOK { + return "", fmt.Errorf("unexpected status code %d from chain %s", statusCode, chainName) } - - // Parse JSON-RPC response var rpcResponse struct { Result string `json:"result"` Error *struct { @@ -1312,20 +1426,44 @@ func (bl *FxBlockchain) callRemoveMemberPeerId(ctx context.Context, poolID uint3 Message string `json:"message"` } `json:"error"` } - if err := json.Unmarshal(response, &rpcResponse); err != nil { - return fmt.Errorf("failed to parse RPC response: %w", err) + return "", fmt.Errorf("failed to parse RPC response: %w", err) } - if rpcResponse.Error != nil { - return fmt.Errorf("RPC error: %s", rpcResponse.Error.Message) + return "", fmt.Errorf("RPC error: %s", rpcResponse.Error.Message) } + return rpcResponse.Result, nil +} - // Check for contract-specific errors - if contractErr := abi.ParseContractError(rpcResponse.Result); contractErr != nil { - return fmt.Errorf("contract error: %w", contractErr) +// ReconcilePoolConfig clears the configured pool when the chain definitively says this device is neither a member +// of it nor waiting to join it — e.g. it was removed from the app while offline, or the app's leave notice never +// arrived — unless the pool was joined from this device less than PoolJoinGracePeriod ago. Any read error keeps the +// config. Returns true when the config was cleared (and the services restart was signalled). +func (bl *FxBlockchain) ReconcilePoolConfig(ctx context.Context) (bool, error) { + if bl.getPoolName == nil || bl.getChainName == nil { + return false, nil } - - log.Infow("Successfully called removeMemberPeerId", "poolID", poolID, "peerID", peerID, "chain", chainName) - return nil + pool, chainName := bl.getPoolName(), bl.getChainName() + if pool == "" || pool == "0" || chainName == "" { + return false, nil + } + if joinedAt, ok := PoolJoinedAt(); ok && time.Since(joinedAt) < PoolJoinGracePeriod { + log.Debugw("Pool reconcile: join is recent, keeping config", "pool", pool, "chain", chainName, "joinedAt", joinedAt) + return false, nil + } + poolID, err := strconv.ParseUint(pool, 10, 32) + if err != nil { + return false, fmt.Errorf("invalid configured pool %q: %w", pool, err) + } + member, pending, err := bl.ClusterPeerPoolStatus(ctx, uint32(poolID), chainName) + if err != nil { + return false, err + } + if member || pending { + return false, nil + } + log.Infow("Pool reconcile: this Blox is no longer in the configured pool on-chain; clearing it", "pool", pool, "chain", chainName) + bl.cleanLeaveJoinPool(ctx, int(poolID)) + bl.signalPoolConfigCleared(0) + return true, nil } diff --git a/blockchain/options.go b/blockchain/options.go index 1fd3790..c4cfd26 100644 --- a/blockchain/options.go +++ b/blockchain/options.go @@ -36,6 +36,7 @@ type ( clusterPeerID peer.ID // IPFS cluster peer ID (original identity), used for on-chain pool membership signingKey crypto.PrivKey // Private key for signing outgoing requests (mobile client) clientProtocolID string // Protocol ID for kubo p2p forwarding (e.g. "/x/fula-blockchain") + onPoolConfigCleared func() // Called after a leave / reconcile cleared the pool from the config } ) @@ -245,6 +246,15 @@ func WithRequestSigning(key crypto.PrivKey) Option { } } +// WithOnPoolConfigCleared sets a hook run after a pool leave or reconcile removed the pool from the config (the +// blox uses it to restart the fula services so ipfs-cluster stops following the old pool). +func WithOnPoolConfigCleared(fn func()) Option { + return func(o *options) error { + o.onPoolConfigCleared = fn + return nil + } +} + // WithClientProtocolID sets the libp2p protocol ID used when dialing through kubo p2p. func WithClientProtocolID(pid string) Option { return func(o *options) error { diff --git a/blockchain/pool_leave_live_test.go b/blockchain/pool_leave_live_test.go new file mode 100644 index 0000000..53aab86 --- /dev/null +++ b/blockchain/pool_leave_live_test.go @@ -0,0 +1,151 @@ +//go:build livechain + +// Read-only checks of the pool-leave path against the real SKALE and Base pool contracts (needs network, so it is +// not part of the normal test run): +// +// go test -tags livechain -run TestLive -v ./blockchain/ +// +// LIVE_MEMBER_PEER_ID / LIVE_CHAIN pick a cluster peer that is a member of pool 1 on that chain (defaults: a SKALE +// pool-1 member). LIVE_OWN_PEER_ID is a cluster peer expected NOT to be in a pool there (default: a fresh random +// key) — on a test Blox, pass the device's own cluster peer id. Nothing here touches a real config or restarts a +// service: the pool-name callbacks and the restart hook are fakes. +package blockchain + +import ( + "crypto/rand" + "errors" + "net/http" + "os" + "testing" + "time" + + "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const defaultLiveMemberPeerID = "12D3KooWMax6UNK5k9Wf1HTRdqJonSmk9khAxC4JUjJMpXFq7w88" // pool 1 on SKALE + +func liveEnv(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +func livePeer(t *testing.T, s string) peer.ID { + t.Helper() + if s == "" { + priv, _, err := crypto.GenerateEd25519Key(rand.Reader) + require.NoError(t, err) + id, err := peer.IDFromPrivateKey(priv) + require.NoError(t, err) + return id + } + id, err := peer.Decode(s) + require.NoError(t, err) + return id +} + +// newLiveBlockchain uses the real chain RPCs (default http client) with fake config callbacks. +func newLiveBlockchain(t *testing.T, clusterPeer peer.ID, cfg *poolConfig) *FxBlockchain { + t.Helper() + PoolJoinedAtFilePath = t.TempDir() + "/pool_joined_at.tmp" + poolLeaveCheckInterval = 2 * time.Second + poolConfigClearedRestartDelay = 0 + bl, err := NewFxBlockchain(NewSimpleKeyStorer(""), + WithTimeout(60), + WithClusterPeerID(clusterPeer), + WithGetPoolName(cfg.getPool), + WithUpdatePoolName(cfg.setPool), + WithGetChainName(cfg.getChain), + WithUpdateChainName(cfg.setChain), + WithOnPoolConfigCleared(func() { + cfg.mu.Lock() + cfg.hookCalls++ + cfg.mu.Unlock() + cfg.cleared <- struct{}{} + }), + ) + require.NoError(t, err) + return bl +} + +func TestLiveClusterPeerPoolStatus(t *testing.T) { + chain := liveEnv("LIVE_CHAIN", "skale") + member := livePeer(t, liveEnv("LIVE_MEMBER_PEER_ID", defaultLiveMemberPeerID)) + own := livePeer(t, os.Getenv("LIVE_OWN_PEER_ID")) + + cfg := &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + isMember, pending, err := newLiveBlockchain(t, member, cfg).ClusterPeerPoolStatus(t.Context(), 1, chain) + require.NoError(t, err) + assert.True(t, isMember, "known member peer %s should be in pool 1 on %s", member, chain) + assert.False(t, pending) + + isMember, pending, err = newLiveBlockchain(t, own, cfg).ClusterPeerPoolStatus(t.Context(), 1, chain) + require.NoError(t, err) + assert.False(t, isMember, "%s should not be in pool 1 on %s", own, chain) + assert.False(t, pending) +} + +func TestLiveHandlePoolLeave(t *testing.T) { + chain := liveEnv("LIVE_CHAIN", "skale") + member := livePeer(t, liveEnv("LIVE_MEMBER_PEER_ID", defaultLiveMemberPeerID)) + own := livePeer(t, os.Getenv("LIVE_OWN_PEER_ID")) + + t.Run("not in the pool on-chain: 202 and config cleared", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + bl := newLiveBlockchain(t, own, cfg) + rec := leave(bl, 1, chain) + assert.Equal(t, http.StatusAccepted, rec.Code, rec.Body.String()) + assert.Equal(t, "0", cfg.getPool()) + assert.Equal(t, "", cfg.getChain()) + waitCleared(t, cfg) + }) + t.Run("still a member on-chain: 409 after re-checks, config kept", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + bl := newLiveBlockchain(t, member, cfg) + rec := leave(bl, 1, chain) + assert.Equal(t, http.StatusConflict, rec.Code, rec.Body.String()) + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, 0, cfg.hookCalls) + }) + t.Run("configured for another pool: 409", func(t *testing.T) { + cfg := &poolConfig{pool: "2", chain: chain, cleared: make(chan struct{}, 1)} + rec := leave(newLiveBlockchain(t, own, cfg), 1, chain) + assert.Equal(t, http.StatusConflict, rec.Code, rec.Body.String()) + assert.Equal(t, "2", cfg.getPool()) + }) + t.Run("chain unreachable: 503, config kept", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + bl := newLiveBlockchain(t, own, cfg) + bl.ch = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("network is unreachable") + })} + rec := leave(bl, 1, chain) + assert.Equal(t, http.StatusServiceUnavailable, rec.Code, rec.Body.String()) + assert.Equal(t, "1", cfg.getPool()) + }) +} + +func TestLiveReconcilePoolConfig(t *testing.T) { + chain := liveEnv("LIVE_CHAIN", "skale") + member := livePeer(t, liveEnv("LIVE_MEMBER_PEER_ID", defaultLiveMemberPeerID)) + own := livePeer(t, os.Getenv("LIVE_OWN_PEER_ID")) + + cfg := &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + cleared, err := newLiveBlockchain(t, member, cfg).ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared, "a real member keeps its pool") + + cfg = &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + cleared, err = newLiveBlockchain(t, own, cfg).ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.True(t, cleared, "a peer that isn't in the pool has it cleared") + waitCleared(t, cfg) +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } diff --git a/blockchain/pool_leave_test.go b/blockchain/pool_leave_test.go new file mode 100644 index 0000000..b59a648 --- /dev/null +++ b/blockchain/pool_leave_test.go @@ -0,0 +1,302 @@ +package blockchain + +import ( + "bytes" + "crypto/rand" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/functionland/go-fula/blockchain/abi" + "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeChain answers the pool contract's eth_calls for every chain RPC (the chain URLs are hard-coded, so the +// http.Client transport is swapped instead). +type fakeChain struct { + mu sync.Mutex + member bool // isPeerIdMemberOfPool result + status uint8 // joinRequests(...).status + failRPC bool // answer every call with a JSON-RPC error + calls int +} + +func (f *fakeChain) RoundTrip(req *http.Request) (*http.Response, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls++ + var body struct { + ID interface{} `json:"id"` + Params json.RawMessage `json:"params"` + } + raw, _ := io.ReadAll(req.Body) + _ = json.Unmarshal(raw, &body) + var params []map[string]string + _ = json.Unmarshal(body.Params, ¶ms) + data := "" + if len(params) > 0 { + data = params[0]["data"] + } + + resp := map[string]interface{}{"jsonrpc": "2.0", "id": body.ID} + word := func(v uint64) string { return fmt.Sprintf("%064x", v) } + switch { + case f.failRPC: + resp["error"] = map[string]interface{}{"code": -32000, "message": "upstream unavailable"} + case strings.HasPrefix(data, abi.MethodSignatures.IsPeerIdMemberOfPool): + isMember := uint64(0) + if f.member { + isMember = 1 + } + resp["result"] = "0x" + word(isMember) + word(0) + case strings.HasPrefix(data, abi.MethodSignatures.JoinRequests): + resp["result"] = "0x" + strings.Repeat(word(0), 6) + word(uint64(f.status)) + word(0) + default: + resp["error"] = map[string]interface{}{"code": -32601, "message": "unexpected call " + data} + } + out, _ := json.Marshal(resp) + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(bytes.NewReader(out)), + Request: req, + }, nil +} + +type poolConfig struct { + mu sync.Mutex + pool string + chain string + cleared chan struct{} + hookCalls int +} + +func (c *poolConfig) getPool() string { c.mu.Lock(); defer c.mu.Unlock(); return c.pool } +func (c *poolConfig) getChain() string { c.mu.Lock(); defer c.mu.Unlock(); return c.chain } +func (c *poolConfig) setPool(p string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.pool = p + return nil +} +func (c *poolConfig) setChain(ch string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.chain = ch + return nil +} + +func newLeaveTestBlockchain(t *testing.T, chain *fakeChain, cfg *poolConfig, withClusterPeer bool) *FxBlockchain { + t.Helper() + PoolJoinedAtFilePath = filepath.Join(t.TempDir(), "pool_joined_at.tmp") + poolLeaveCheckInterval = 10 * time.Millisecond + poolConfigClearedRestartDelay = 0 + + opts := []Option{ + WithTimeout(30), + WithGetPoolName(cfg.getPool), + WithUpdatePoolName(cfg.setPool), + WithGetChainName(cfg.getChain), + WithUpdateChainName(cfg.setChain), + WithOnPoolConfigCleared(func() { + cfg.mu.Lock() + cfg.hookCalls++ + cfg.mu.Unlock() + cfg.cleared <- struct{}{} + }), + } + if withClusterPeer { + priv, _, err := crypto.GenerateEd25519Key(rand.Reader) + require.NoError(t, err) + id, err := peer.IDFromPrivateKey(priv) + require.NoError(t, err) + opts = append(opts, WithClusterPeerID(id)) + } + bl, err := NewFxBlockchain(NewSimpleKeyStorer(""), opts...) + require.NoError(t, err) + bl.ch = &http.Client{Transport: chain} + return bl +} + +func leave(bl *FxBlockchain, poolID int, chainName string) *httptest.ResponseRecorder { + body := fmt.Sprintf(`{"pool_id":%d,"chain_name":%q}`, poolID, chainName) + req := httptest.NewRequest(http.MethodPost, "/"+actionPoolLeave, strings.NewReader(body)) + rec := httptest.NewRecorder() + bl.HandlePoolLeave(http.MethodPost, actionPoolLeave, "", rec, req) + return rec +} + +func waitCleared(t *testing.T, cfg *poolConfig) { + t.Helper() + select { + case <-cfg.cleared: + case <-time.After(2 * time.Second): + t.Fatal("pool-config-cleared hook was not called") + } +} + +func TestHandlePoolLeave_NotMemberClearsConfigAndRestarts(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + recordPoolJoinTime(time.Now()) // a leave clears even a fresh join + + rec := leave(bl, 1, "skale") + + assert.Equal(t, http.StatusAccepted, rec.Code, rec.Body.String()) + var res PoolLeaveResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &res)) + assert.Equal(t, 1, res.PoolID) + assert.Equal(t, "skale", res.ChainName) + assert.Equal(t, "0", cfg.getPool()) + assert.Equal(t, "", cfg.getChain()) + assert.Equal(t, "0", bl.topicName) + _, ok := PoolJoinedAt() + assert.False(t, ok, "join marker is removed on leave") + waitCleared(t, cfg) +} + +func TestHandlePoolLeave_StillMemberOrPendingIs409(t *testing.T) { + for name, chain := range map[string]*fakeChain{ + "member": {member: true}, + "pending": {status: 1}, + } { + t.Run(name, func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "base", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + + rec := leave(bl, 1, "base") + + assert.Equal(t, http.StatusConflict, rec.Code, rec.Body.String()) + assert.Contains(t, rec.Body.String(), "remove it on-chain first") + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, "base", cfg.getChain()) + assert.Equal(t, 0, cfg.hookCalls) + }) + } +} + +func TestHandlePoolLeave_RechecksBeforeGivingUp(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "base", cleared: make(chan struct{}, 1)} + chain := &fakeChain{member: true} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + + rec := leave(bl, 1, "base") + + assert.Equal(t, http.StatusConflict, rec.Code) + assert.Equal(t, poolLeaveCheckAttempts, chain.calls, "one isPeerIdMemberOfPool read per attempt") +} + +func TestHandlePoolLeave_OtherConfiguredPoolIs409WithoutRPC(t *testing.T) { + cfg := &poolConfig{pool: "2", chain: "skale", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + + rec := leave(bl, 1, "skale") + + assert.Equal(t, http.StatusConflict, rec.Code) + assert.Contains(t, rec.Body.String(), "configured for pool 2") + assert.Equal(t, 0, chain.calls) + assert.Equal(t, "2", cfg.getPool()) +} + +func TestHandlePoolLeave_UnverifiableIs503AndKeepsConfig(t *testing.T) { + t.Run("rpc error", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{failRPC: true}, cfg, true) + rec := leave(bl, 1, "skale") + assert.Equal(t, http.StatusServiceUnavailable, rec.Code) + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, 0, cfg.hookCalls) + }) + t.Run("no cluster peer id", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, false) + rec := leave(bl, 1, "skale") + assert.Equal(t, http.StatusServiceUnavailable, rec.Code) + assert.Equal(t, "1", cfg.getPool()) + }) +} + +func TestReconcilePoolConfig(t *testing.T) { + t.Run("not a member and no join marker: cleared", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.True(t, cleared) + assert.Equal(t, "0", cfg.getPool()) + assert.Equal(t, "", cfg.getChain()) + waitCleared(t, cfg) + }) + t.Run("recent join: kept without reading the chain", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + recordPoolJoinTime(time.Now().Add(-time.Hour)) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, 0, chain.calls) + }) + t.Run("join older than the grace period: cleared", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "base", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + recordPoolJoinTime(time.Now().Add(-PoolJoinGracePeriod - time.Minute)) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.True(t, cleared) + waitCleared(t, cfg) + }) + t.Run("still a member or pending: kept", func(t *testing.T) { + for _, chain := range []*fakeChain{{member: true}, {status: 1}} { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + } + }) + t.Run("rpc error: kept, error returned", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{failRPC: true}, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + assert.Error(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, 0, cfg.hookCalls) + }) + t.Run("no pool configured: nothing to do", func(t *testing.T) { + cfg := &poolConfig{pool: "0", chain: "", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, 0, chain.calls) + }) +} + +func TestHandlePoolJoinRecordsJoinTime(t *testing.T) { + PoolJoinedAtFilePath = filepath.Join(t.TempDir(), "pool_joined_at.tmp") + before := time.Now().Add(-time.Second) + recordPoolJoinTime(time.Now()) + at, ok := PoolJoinedAt() + require.True(t, ok) + assert.True(t, at.After(before)) + clearPoolJoinTime() + _, err := os.Stat(PoolJoinedAtFilePath) + assert.True(t, os.IsNotExist(err)) +} diff --git a/blox/blox.go b/blox/blox.go index 119c5b5..47a6bbd 100644 --- a/blox/blox.go +++ b/blox/blox.go @@ -96,6 +96,13 @@ func New(o ...Option) (*Blox, error) { blockchain.WithIpfsClient(p.rpc), blockchain.WithMinSuccessPingCount(p.minSuccessRate*p.pingCount/100), blockchain.WithIpfsClusterAPI(p.ipfsClusterApi), + // After a pool leave / reconcile cleared the config, restart the fula services so ipfs-cluster re-reads + // the (now empty) pool instead of following the old one until the next reboot. + blockchain.WithOnPoolConfigCleared(func() { + if err := signalFulaRestart(); err != nil { + log.Errorw("Failed to signal fula restart after the pool was cleared", "err", err) + } + }), ) if err != nil { return nil, err @@ -427,6 +434,10 @@ func (p *Blox) Start(ctx context.Context) error { p.watchKuboP2P(ctx) }() + // Drop a configured pool this Blox no longer belongs to on-chain (left from the app while it was offline, or the + // leave notice never arrived); discovery below then runs as for a fresh device. + p.reconcilePool(ctx) + // Register cluster tunnel forward immediately if pool is already known from config, // before potentially blocking on chain discovery for up to ~15 minutes. if p.topicName != "0" { @@ -637,6 +648,7 @@ func (p *Blox) Start(ctx context.Context) error { for { select { case <-ticker.C: + p.reconcilePool(p.ctx) // instead of FetchAvailableManifestsAndStore See what are the new manifests from ther last time we checked in blockstore // A method that checks the last time we checked for stored blocks file // then it checks the stored files under /uniondrive/ipfs_datastore/blocks/{folders that are not .temp} diff --git a/blox/pool_reconcile.go b/blox/pool_reconcile.go new file mode 100644 index 0000000..0870d0e --- /dev/null +++ b/blox/pool_reconcile.go @@ -0,0 +1,25 @@ +package blox + +import ( + "context" + "time" +) + +// poolReconcileTimeout bounds one reconcile pass (two chain reads, each with retries). +const poolReconcileTimeout = 60 * time.Second + +// reconcilePool clears the configured pool when the chain definitively says this Blox left it (see +// blockchain.ReconcilePoolConfig). Read errors keep the config; the next pass (startup or the 6 h loop) retries. +func (p *Blox) reconcilePool(ctx context.Context) { + ctx, cancel := context.WithTimeout(ctx, poolReconcileTimeout) + defer cancel() + cleared, err := p.bl.ReconcilePoolConfig(ctx) + if err != nil { + log.Warnw("Pool reconcile skipped; keeping the configured pool", "err", err) + return + } + if cleared { + p.topicName = "0" + p.chainName = "" + } +} From cea20e94590c3511189494016975b6c3a4859d9a Mon Sep 17 00:00:00 2001 From: ehsan shariati Date: Mon, 5 Oct 2026 01:24:57 -0400 Subject: [PATCH 2/6] fix(blockchain): pool leave refuses another chain; no-op when unconfigured - A leave for pool N on chain B must not clear a config of pool N on chain A (pool ids repeat across chains; some Bloxes are in pool 1 on both SKALE and Base): answer 409 "Chain mismatch". - With no pool configured (e.g. a reconcile already cleared it), answer 202 straight away, without reading the chain or restarting services. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QaKz9HeXaNBjrchbbGimFx --- blockchain/bl_pool.go | 32 +++++++++++++++++++++++++------- blockchain/pool_leave_test.go | 28 ++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 7 deletions(-) diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index 0d5b939..4047707 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -1194,7 +1194,9 @@ func (bl *FxBlockchain) validatePoolOnChain(ctx context.Context, poolID uint32, // cluster peer is neither a member of the pool nor waiting to join it: // - 202: removed on-chain (or never there) → pool and chain cleared from config, fula services restarted so // ipfs-cluster leaves the old pool; -// - 409: the device is configured for a different pool, or the chain still lists it (the app's tx not mined yet); +// - 202 without changes when no pool is configured; +// - 409: the device is configured for a different pool or chain, or the chain still lists it (the app's tx not +// mined yet); // - 503: the chain could not be read, or the cluster peer id is unknown — config untouched, safe to retry. func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer.ID, w http.ResponseWriter, r *http.Request) { log := log.With("action", action, "from", from) @@ -1230,15 +1232,24 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. fmt.Sprintf("This Blox is configured for pool %s, not pool %d", configuredPool, poolID)) return } + // Nothing configured (e.g. a reconcile already cleared it): nothing to clean up or restart. + if configuredPool == "" || configuredPool == "0" { + log.Infow("Pool leave: no pool configured, nothing to do", "poolID", poolID, "chain", chainName) + w.WriteHeader(http.StatusAccepted) + if err := json.NewEncoder(w).Encode(PoolLeaveResponse{PoolID: req.PoolID, ChainName: chainName}); err != nil { + log.Error("failed to write response: %v", err) + } + return + } + configuredChain := "" + if bl.getChainName != nil { + configuredChain = bl.getChainName() + } // If no chain name provided, try to determine it from current configuration if chainName == "" { - var currentChain string - if bl.getChainName != nil { - currentChain = bl.getChainName() - } - if currentChain != "" { - chainName = currentChain + if configuredChain != "" { + chainName = configuredChain log.Debugw("Using current chain configuration", "poolID", poolID, "chain", chainName) } else { // Try to discover which chain this pool exists on @@ -1253,6 +1264,13 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. } } } + // Pool ids repeat across chains (a Blox can even be in pool 1 on both): a leave on another chain than the one + // this Blox is configured for must not clear its config. + if configuredChain != "" && configuredChain != chainName { + writeError(http.StatusConflict, "Chain mismatch", + fmt.Sprintf("This Blox is configured for pool %s on %s, not on %s", configuredPool, configuredChain, chainName)) + return + } // The app sends this right after its removeMemberPeerId tx is mined; a public RPC can lag a block or two. stillListed := false diff --git a/blockchain/pool_leave_test.go b/blockchain/pool_leave_test.go index b59a648..9a7b5e4 100644 --- a/blockchain/pool_leave_test.go +++ b/blockchain/pool_leave_test.go @@ -210,6 +210,34 @@ func TestHandlePoolLeave_OtherConfiguredPoolIs409WithoutRPC(t *testing.T) { assert.Equal(t, "2", cfg.getPool()) } +func TestHandlePoolLeave_OtherConfiguredChainIs409WithoutRPC(t *testing.T) { + // Pool 1 exists on both chains; a leave of Base pool 1 must not clear a SKALE pool-1 config. + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + + rec := leave(bl, 1, "base") + + assert.Equal(t, http.StatusConflict, rec.Code) + assert.Contains(t, rec.Body.String(), "configured for pool 1 on skale, not on base") + assert.Equal(t, 0, chain.calls) + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, "skale", cfg.getChain()) +} + +func TestHandlePoolLeave_NothingConfiguredIs202WithoutRestart(t *testing.T) { + cfg := &poolConfig{pool: "0", chain: "", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + + rec := leave(bl, 1, "skale") + + assert.Equal(t, http.StatusAccepted, rec.Code) + assert.Equal(t, 0, chain.calls) + time.Sleep(50 * time.Millisecond) + assert.Equal(t, 0, cfg.hookCalls, "no services restart when there was nothing to clear") +} + func TestHandlePoolLeave_UnverifiableIs503AndKeepsConfig(t *testing.T) { t.Run("rpc error", func(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} From b5558ea2e3e633f04447c1a4adac910bd75e71a5 Mon Sep 17 00:00:00 2001 From: ehsan shariati Date: Mon, 5 Oct 2026 01:40:24 -0400 Subject: [PATCH 3/6] fix(blox): keep pool reconcile safe for running devices - Run the startup reconcile in the background 2 min after start (off the startup path, after the network settles), and never on pool hosts (--poolHost), whose own pool is not a chain-listed membership. - Only signal the fula services restart after the pool is confirmed gone from the config: if the config write failed, a restart would find the pool again and loop. Leave answers 500 in that case. - Don't write Blox fields from the background reconcile; the restart reloads them. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QaKz9HeXaNBjrchbbGimFx --- blockchain/bl_pool.go | 23 +++++++++++++++++++++-- blockchain/pool_leave_test.go | 35 ++++++++++++++++++++++++++++++----- blox/blox.go | 5 +++-- blox/pool_reconcile.go | 32 +++++++++++++++++++++++++++----- 4 files changed, 81 insertions(+), 14 deletions(-) diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index 4047707..f4886c1 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -1300,7 +1300,11 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. } // Clean up local configuration and state - bl.cleanLeaveJoinPool(ctx, poolID) + if err := bl.clearPoolConfig(ctx, poolID); err != nil { + log.Errorw("Pool leave: could not clear the pool from the config", "poolID", poolID, "error", err) + writeError(http.StatusInternalServerError, "Failed to update the Blox config", err.Error()) + return + } statusCode := http.StatusAccepted res = PoolLeaveResponse{ @@ -1481,7 +1485,22 @@ func (bl *FxBlockchain) ReconcilePoolConfig(ctx context.Context) (bool, error) { return false, nil } log.Infow("Pool reconcile: this Blox is no longer in the configured pool on-chain; clearing it", "pool", pool, "chain", chainName) - bl.cleanLeaveJoinPool(ctx, int(poolID)) + if err := bl.clearPoolConfig(ctx, int(poolID)); err != nil { + return false, err + } bl.signalPoolConfigCleared(0) return true, nil } + +// clearPoolConfig drops the pool from the config (cleanLeaveJoinPool) and confirms it took effect. The services are +// only restarted after a confirmed clear: if the config write failed, a restart would find the pool again, clear +// and restart again — a loop. +func (bl *FxBlockchain) clearPoolConfig(ctx context.Context, poolID int) error { + bl.cleanLeaveJoinPool(ctx, poolID) + if bl.getPoolName != nil { + if pool := bl.getPoolName(); pool != "" && pool != "0" { + return fmt.Errorf("pool is still %q in the config after clearing it", pool) + } + } + return nil +} diff --git a/blockchain/pool_leave_test.go b/blockchain/pool_leave_test.go index 9a7b5e4..72a01a7 100644 --- a/blockchain/pool_leave_test.go +++ b/blockchain/pool_leave_test.go @@ -75,11 +75,12 @@ func (f *fakeChain) RoundTrip(req *http.Request) (*http.Response, error) { } type poolConfig struct { - mu sync.Mutex - pool string - chain string - cleared chan struct{} - hookCalls int + mu sync.Mutex + pool string + chain string + cleared chan struct{} + hookCalls int + failWrites bool // simulate a config.yaml that can't be written } func (c *poolConfig) getPool() string { c.mu.Lock(); defer c.mu.Unlock(); return c.pool } @@ -87,6 +88,9 @@ func (c *poolConfig) getChain() string { c.mu.Lock(); defer c.mu.Unlock(); retur func (c *poolConfig) setPool(p string) error { c.mu.Lock() defer c.mu.Unlock() + if c.failWrites { + return fmt.Errorf("read-only file system") + } c.pool = p return nil } @@ -238,6 +242,27 @@ func TestHandlePoolLeave_NothingConfiguredIs202WithoutRestart(t *testing.T) { assert.Equal(t, 0, cfg.hookCalls, "no services restart when there was nothing to clear") } +func TestConfigWriteFailureNeverRestarts(t *testing.T) { + // A restart after a failed write would find the pool again, clear, restart again — a loop. + t.Run("leave", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1), failWrites: true} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + rec := leave(bl, 1, "skale") + assert.Equal(t, http.StatusInternalServerError, rec.Code, rec.Body.String()) + time.Sleep(50 * time.Millisecond) + assert.Equal(t, 0, cfg.hookCalls) + }) + t.Run("reconcile", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1), failWrites: true} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + assert.Error(t, err) + assert.False(t, cleared) + time.Sleep(50 * time.Millisecond) + assert.Equal(t, 0, cfg.hookCalls) + }) +} + func TestHandlePoolLeave_UnverifiableIs503AndKeepsConfig(t *testing.T) { t.Run("rpc error", func(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} diff --git a/blox/blox.go b/blox/blox.go index 47a6bbd..288464c 100644 --- a/blox/blox.go +++ b/blox/blox.go @@ -435,8 +435,9 @@ func (p *Blox) Start(ctx context.Context) error { }() // Drop a configured pool this Blox no longer belongs to on-chain (left from the app while it was offline, or the - // leave notice never arrived); discovery below then runs as for a fresh device. - p.reconcilePool(ctx) + // leave notice never arrived). Runs in the background a little after start, never on pool hosts; a clear + // restarts the fula services, after which discovery runs as for a fresh device. + p.startPoolReconcile(ctx) // Register cluster tunnel forward immediately if pool is already known from config, // before potentially blocking on chain discovery for up to ~15 minutes. diff --git a/blox/pool_reconcile.go b/blox/pool_reconcile.go index 0870d0e..6625a1f 100644 --- a/blox/pool_reconcile.go +++ b/blox/pool_reconcile.go @@ -5,11 +5,16 @@ import ( "time" ) -// poolReconcileTimeout bounds one reconcile pass (two chain reads, each with retries). -const poolReconcileTimeout = 60 * time.Second +const ( + // poolReconcileTimeout bounds one reconcile pass (two chain reads, each with retries). + poolReconcileTimeout = 60 * time.Second + // poolReconcileStartupDelay keeps the first pass off the startup path and lets the network settle after a boot. + poolReconcileStartupDelay = 2 * time.Minute +) // reconcilePool clears the configured pool when the chain definitively says this Blox left it (see -// blockchain.ReconcilePoolConfig). Read errors keep the config; the next pass (startup or the 6 h loop) retries. +// blockchain.ReconcilePoolConfig); the services restart that follows reloads the pool state. Read errors keep the +// config; the next pass (the 6 h loop) retries. func (p *Blox) reconcilePool(ctx context.Context) { ctx, cancel := context.WithTimeout(ctx, poolReconcileTimeout) defer cancel() @@ -19,7 +24,24 @@ func (p *Blox) reconcilePool(ctx context.Context) { return } if cleared { - p.topicName = "0" - p.chainName = "" + log.Info("Pool reconcile cleared the configured pool; fula services restart requested") + } +} + +// startPoolReconcile runs the first reconcile pass in the background, poolReconcileStartupDelay after start. Pool +// hosts are skipped: their own pool is not a membership the chain lists for them. +func (p *Blox) startPoolReconcile(ctx context.Context) { + if p.poolHostMode { + return } + p.wg.Add(1) + go func() { + defer p.wg.Done() + select { + case <-ctx.Done(): + return + case <-time.After(poolReconcileStartupDelay): + } + p.reconcilePool(ctx) + }() } From 0a862b40406b9ee4683a75fd291b5146472525af Mon Sep 17 00:00:00 2001 From: ehsan shariati Date: Tue, 6 Oct 2026 01:03:31 -0400 Subject: [PATCH 4/6] fix(blockchain): bound pool reconcile so it can never loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - At most one automatic (reconcile) clear — and services restart — per 24 h, recorded in /internal/.tmp/pool_reconcile_cleared_at.tmp before clearing; if that record can't be written, nothing is cleared. - Act only on two definitive "not listed" answers 30 s apart, so a single lagging RPC node can't clear a member's pool. - clearPoolConfig checks the config write errors itself instead of trusting a re-read that falls back to "0" when config.yaml can't be read. - Don't write in-memory topic/chain names from the leave / reconcile goroutines (other request goroutines read them; the restart reloads them). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QaKz9HeXaNBjrchbbGimFx --- blockchain/bl_pool.go | 96 ++++++++++++++++++++++-------- blockchain/pool_leave_live_test.go | 5 +- blockchain/pool_leave_test.go | 69 ++++++++++++++++++--- blox/pool_reconcile.go | 4 +- 4 files changed, 137 insertions(+), 37 deletions(-) diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index f4886c1..dba7a82 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -328,8 +328,8 @@ func (bl *FxBlockchain) cleanLeaveJoinPool(ctx context.Context, PoolID int) { log.Errorw("Failed to reset chain name", "error", err) } } - bl.topicName = "0" - bl.chainName = "" + // (In-memory topic/chain names are not touched here — other request goroutines read them; the services restart + // that follows a clear reloads them from the config.) clearPoolJoinTime() bl.StopPingServer(ctx) @@ -1329,25 +1329,29 @@ const ( // the Blox yet: a join writes the config immediately, but the join server's addMember (or approval votes for a // contract join) lands later. PoolJoinGracePeriod = 24 * time.Hour - joinRequestPending = 1 // JoinRequest.status of a request still waiting for votes + // PoolReconcileMinInterval caps automatic (reconcile) clears — and the services restart each one triggers — to + // one per interval, so no combination of chain / RPC answers can make a restart loop. + PoolReconcileMinInterval = 24 * time.Hour + joinRequestPending = 1 // JoinRequest.status of a request still waiting for votes ) // Variables (not constants) so tests can shorten them. var ( poolLeaveCheckInterval = 4 * time.Second poolConfigClearedRestartDelay = 3 * time.Second + poolReconcileConfirmDelay = 30 * time.Second ) -// PoolJoinedAtFilePath records when HandlePoolJoin last wrote a pool into the config (see PoolJoinGracePeriod). -// A variable so tests can point it at a temp dir. -var PoolJoinedAtFilePath = "/internal/.tmp/pool_joined_at.tmp" +// Marker files (variables so tests can point them at a temp dir). PoolJoinedAtFilePath records when HandlePoolJoin +// last wrote a pool into the config (see PoolJoinGracePeriod); PoolReconcileClearedAtFilePath when +// ReconcilePoolConfig last cleared one (see PoolReconcileMinInterval). +var ( + PoolJoinedAtFilePath = "/internal/.tmp/pool_joined_at.tmp" + PoolReconcileClearedAtFilePath = "/internal/.tmp/pool_reconcile_cleared_at.tmp" +) func recordPoolJoinTime(t time.Time) { - if err := os.MkdirAll(filepath.Dir(PoolJoinedAtFilePath), 0755); err != nil { - log.Warnw("Failed to create directory for pool join time", "err", err) - return - } - if err := os.WriteFile(PoolJoinedAtFilePath, []byte(t.UTC().Format(time.RFC3339)), 0644); err != nil { + if err := writeTimeMarker(PoolJoinedAtFilePath, t); err != nil { log.Warnw("Failed to record pool join time", "err", err) } } @@ -1361,15 +1365,7 @@ func clearPoolJoinTime() { // PoolJoinedAt returns when the configured pool was joined from this device; false when unknown (e.g. joins made // before this marker existed), which counts as an old join. func PoolJoinedAt() (time.Time, bool) { - b, err := os.ReadFile(PoolJoinedAtFilePath) - if err != nil { - return time.Time{}, false - } - t, err := time.Parse(time.RFC3339, strings.TrimSpace(string(b))) - if err != nil { - return time.Time{}, false - } - return t, true + return readTimeMarker(PoolJoinedAtFilePath) } // signalPoolConfigCleared runs the WithOnPoolConfigCleared hook (the blox wires it to a fula services restart, so @@ -1473,16 +1469,35 @@ func (bl *FxBlockchain) ReconcilePoolConfig(ctx context.Context) (bool, error) { log.Debugw("Pool reconcile: join is recent, keeping config", "pool", pool, "chain", chainName, "joinedAt", joinedAt) return false, nil } + // At most one reconcile clear (and services restart) per PoolReconcileMinInterval, whatever the RPCs answer. + if clearedAt, ok := readTimeMarker(PoolReconcileClearedAtFilePath); ok && time.Since(clearedAt) < PoolReconcileMinInterval { + log.Debugw("Pool reconcile: cleared recently, skipping", "clearedAt", clearedAt) + return false, nil + } poolID, err := strconv.ParseUint(pool, 10, 32) if err != nil { return false, fmt.Errorf("invalid configured pool %q: %w", pool, err) } - member, pending, err := bl.ClusterPeerPoolStatus(ctx, uint32(poolID), chainName) - if err != nil { - return false, err + // Two definitive "not listed" answers, a little apart, before acting (RPC nodes can briefly disagree). + for check := 1; check <= 2; check++ { + if check == 2 { + select { + case <-ctx.Done(): + return false, ctx.Err() + case <-time.After(poolReconcileConfirmDelay): + } + } + member, pending, err := bl.ClusterPeerPoolStatus(ctx, uint32(poolID), chainName) + if err != nil { + return false, err + } + if member || pending { + return false, nil + } } - if member || pending { - return false, nil + // Record the clear before doing it: if the record can't be written, the rate limit couldn't hold, so don't clear. + if err := writeTimeMarker(PoolReconcileClearedAtFilePath, time.Now()); err != nil { + return false, fmt.Errorf("cannot record the reconcile time, not clearing: %w", err) } log.Infow("Pool reconcile: this Blox is no longer in the configured pool on-chain; clearing it", "pool", pool, "chain", chainName) if err := bl.clearPoolConfig(ctx, int(poolID)); err != nil { @@ -1496,7 +1511,17 @@ func (bl *FxBlockchain) ReconcilePoolConfig(ctx context.Context) (bool, error) { // only restarted after a confirmed clear: if the config write failed, a restart would find the pool again, clear // and restart again — a loop. func (bl *FxBlockchain) clearPoolConfig(ctx context.Context, poolID int) error { - bl.cleanLeaveJoinPool(ctx, poolID) + if bl.updatePoolName != nil { + if err := bl.updatePoolName("0"); err != nil { + return fmt.Errorf("failed to clear the pool in the config: %w", err) + } + } + if bl.updateChainName != nil { + if err := bl.updateChainName(""); err != nil { + return fmt.Errorf("failed to clear the chain in the config: %w", err) + } + } + bl.cleanLeaveJoinPool(ctx, poolID) // the config writes above make its own resets no-ops if bl.getPoolName != nil { if pool := bl.getPoolName(); pool != "" && pool != "0" { return fmt.Errorf("pool is still %q in the config after clearing it", pool) @@ -1504,3 +1529,22 @@ func (bl *FxBlockchain) clearPoolConfig(ctx context.Context, poolID int) error { } return nil } + +func readTimeMarker(path string) (time.Time, bool) { + b, err := os.ReadFile(path) + if err != nil { + return time.Time{}, false + } + t, err := time.Parse(time.RFC3339, strings.TrimSpace(string(b))) + if err != nil { + return time.Time{}, false + } + return t, true +} + +func writeTimeMarker(path string, t time.Time) error { + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { + return err + } + return os.WriteFile(path, []byte(t.UTC().Format(time.RFC3339)), 0644) +} diff --git a/blockchain/pool_leave_live_test.go b/blockchain/pool_leave_live_test.go index 53aab86..6ccf1c0 100644 --- a/blockchain/pool_leave_live_test.go +++ b/blockchain/pool_leave_live_test.go @@ -51,9 +51,12 @@ func livePeer(t *testing.T, s string) peer.ID { // newLiveBlockchain uses the real chain RPCs (default http client) with fake config callbacks. func newLiveBlockchain(t *testing.T, clusterPeer peer.ID, cfg *poolConfig) *FxBlockchain { t.Helper() - PoolJoinedAtFilePath = t.TempDir() + "/pool_joined_at.tmp" + dir := t.TempDir() + PoolJoinedAtFilePath = dir + "/pool_joined_at.tmp" + PoolReconcileClearedAtFilePath = dir + "/pool_reconcile_cleared_at.tmp" poolLeaveCheckInterval = 2 * time.Second poolConfigClearedRestartDelay = 0 + poolReconcileConfirmDelay = 2 * time.Second bl, err := NewFxBlockchain(NewSimpleKeyStorer(""), WithTimeout(60), WithClusterPeerID(clusterPeer), diff --git a/blockchain/pool_leave_test.go b/blockchain/pool_leave_test.go index 72a01a7..d727206 100644 --- a/blockchain/pool_leave_test.go +++ b/blockchain/pool_leave_test.go @@ -25,11 +25,12 @@ import ( // fakeChain answers the pool contract's eth_calls for every chain RPC (the chain URLs are hard-coded, so the // http.Client transport is swapped instead). type fakeChain struct { - mu sync.Mutex - member bool // isPeerIdMemberOfPool result - status uint8 // joinRequests(...).status - failRPC bool // answer every call with a JSON-RPC error - calls int + mu sync.Mutex + member bool // isPeerIdMemberOfPool result + memberAnswers []bool // if set, consumed one per isPeerIdMemberOfPool call (then `member`) + status uint8 // joinRequests(...).status + failRPC bool // answer every call with a JSON-RPC error + calls int } func (f *fakeChain) RoundTrip(req *http.Request) (*http.Response, error) { @@ -55,8 +56,12 @@ func (f *fakeChain) RoundTrip(req *http.Request) (*http.Response, error) { case f.failRPC: resp["error"] = map[string]interface{}{"code": -32000, "message": "upstream unavailable"} case strings.HasPrefix(data, abi.MethodSignatures.IsPeerIdMemberOfPool): + answer := f.member + if len(f.memberAnswers) > 0 { + answer, f.memberAnswers = f.memberAnswers[0], f.memberAnswers[1:] + } isMember := uint64(0) - if f.member { + if answer { isMember = 1 } resp["result"] = "0x" + word(isMember) + word(0) @@ -103,9 +108,12 @@ func (c *poolConfig) setChain(ch string) error { func newLeaveTestBlockchain(t *testing.T, chain *fakeChain, cfg *poolConfig, withClusterPeer bool) *FxBlockchain { t.Helper() - PoolJoinedAtFilePath = filepath.Join(t.TempDir(), "pool_joined_at.tmp") + dir := t.TempDir() + PoolJoinedAtFilePath = filepath.Join(dir, "pool_joined_at.tmp") + PoolReconcileClearedAtFilePath = filepath.Join(dir, "pool_reconcile_cleared_at.tmp") poolLeaveCheckInterval = 10 * time.Millisecond poolConfigClearedRestartDelay = 0 + poolReconcileConfirmDelay = 0 opts := []Option{ WithTimeout(30), @@ -164,7 +172,6 @@ func TestHandlePoolLeave_NotMemberClearsConfigAndRestarts(t *testing.T) { assert.Equal(t, "skale", res.ChainName) assert.Equal(t, "0", cfg.getPool()) assert.Equal(t, "", cfg.getChain()) - assert.Equal(t, "0", bl.topicName) _, ok := PoolJoinedAt() assert.False(t, ok, "join marker is removed on leave") waitCleared(t, cfg) @@ -342,6 +349,52 @@ func TestReconcilePoolConfig(t *testing.T) { }) } +func TestReconcilePoolConfigLoopGuards(t *testing.T) { + t.Run("at most one clear per interval: a second pass within 24 h does nothing", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + require.True(t, cleared) + waitCleared(t, cfg) + + // Something (e.g. discovery after the restart) puts a pool back; the chain still doesn't list the Blox. + require.NoError(t, cfg.setPool("1")) + require.NoError(t, cfg.setChain("skale")) + calls := chain.calls + cleared, err = bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, calls, chain.calls, "no chain reads while rate-limited") + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, 1, cfg.hookCalls) + }) + t.Run("a second read that disagrees keeps the config", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{memberAnswers: []bool{false, true}}, cfg, true) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + _, recorded := readTimeMarker(PoolReconcileClearedAtFilePath) + assert.False(t, recorded) + }) + t.Run("if the clear can't be recorded, nothing is cleared", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + blocker := filepath.Join(t.TempDir(), "not-a-dir") + require.NoError(t, os.WriteFile(blocker, []byte("x"), 0644)) + PoolReconcileClearedAtFilePath = filepath.Join(blocker, "marker.tmp") // parent is a file → unwritable + cleared, err := bl.ReconcilePoolConfig(t.Context()) + assert.Error(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + time.Sleep(50 * time.Millisecond) + assert.Equal(t, 0, cfg.hookCalls) + }) +} + func TestHandlePoolJoinRecordsJoinTime(t *testing.T) { PoolJoinedAtFilePath = filepath.Join(t.TempDir(), "pool_joined_at.tmp") before := time.Now().Add(-time.Second) diff --git a/blox/pool_reconcile.go b/blox/pool_reconcile.go index 6625a1f..1ee083c 100644 --- a/blox/pool_reconcile.go +++ b/blox/pool_reconcile.go @@ -6,8 +6,8 @@ import ( ) const ( - // poolReconcileTimeout bounds one reconcile pass (two chain reads, each with retries). - poolReconcileTimeout = 60 * time.Second + // poolReconcileTimeout bounds one reconcile pass (two confirmed rounds of chain reads, each with retries). + poolReconcileTimeout = 3 * time.Minute // poolReconcileStartupDelay keeps the first pass off the startup path and lets the network settle after a boot. poolReconcileStartupDelay = 2 * time.Minute ) From 2f9ff67b76e1fa749e1d6f125b29eed90be4b808 Mon Sep 17 00:00:00 2001 From: ehsan shariati Date: Tue, 6 Oct 2026 01:32:07 -0400 Subject: [PATCH 5/6] fix(blox): reconcile only clears a pool the Blox was seen leaving MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit So that the update changes nothing for devices as they run today: - ReconcilePoolConfig only clears a configured pool this firmware has itself seen the Blox be a member of (recorded in /internal/.tmp/pool_member_confirmed.tmp as ":") and the chain now says it isn't. A configured pool never seen as a membership — e.g. a join whose on-chain part never landed — is left alone, as before. Replaces the 24 h join-time grace marker. - Re-read the config just before clearing and abort if it changed (a join or rejoin landing during the 30 s check is not wiped). - Start the reconcile goroutine only after startup pool discovery, and serialise the config.yaml accessors in cmd/blox with a mutex, so concurrent read-modify-writes can't mix fields. - Kill switch: while /internal/.disable_pool_reconcile exists, reconcile does nothing. - Pool hosts (--poolHost) refuse pool leave requests (409) and never reconcile. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QaKz9HeXaNBjrchbbGimFx --- blockchain/bl_pool.go | 96 +++++++++++++------ blockchain/options.go | 9 ++ blockchain/pool_leave_live_test.go | 15 ++- blockchain/pool_leave_test.go | 147 +++++++++++++++++++++-------- blox/blox.go | 11 ++- cmd/blox/main.go | 15 +++ 6 files changed, 217 insertions(+), 76 deletions(-) diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index dba7a82..58fdc50 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -131,8 +131,6 @@ func (bl *FxBlockchain) HandlePoolJoin(method string, action string, from peer.I return } - recordPoolJoinTime(time.Now()) - statusCode := http.StatusAccepted res = PoolJoinResponse{ Account: "", @@ -330,7 +328,7 @@ func (bl *FxBlockchain) cleanLeaveJoinPool(ctx context.Context, PoolID int) { } // (In-memory topic/chain names are not touched here — other request goroutines read them; the services restart // that follows a clear reloads them from the config.) - clearPoolJoinTime() + clearPoolMemberConfirmed() bl.StopPingServer(ctx) // Announcements no longer used - pool joins handled via blockchain API @@ -1222,6 +1220,11 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. poolID := req.PoolID chainName := req.ChainName + if bl.poolHost { + writeError(http.StatusConflict, "Pool host", "A pool host does not leave its pool") + return + } + // Never clear a different pool than the one asked about. configuredPool := "0" if bl.getPoolName != nil { @@ -1325,10 +1328,6 @@ func (bl *FxBlockchain) HandlePoolLeave(method string, action string, from peer. const ( poolLeaveCheckAttempts = 3 - // PoolJoinGracePeriod is how long after a join the configured pool is kept even though the chain doesn't list - // the Blox yet: a join writes the config immediately, but the join server's addMember (or approval votes for a - // contract join) lands later. - PoolJoinGracePeriod = 24 * time.Hour // PoolReconcileMinInterval caps automatic (reconcile) clears — and the services restart each one triggers — to // one per interval, so no combination of chain / RPC answers can make a restart loop. PoolReconcileMinInterval = 24 * time.Hour @@ -1342,30 +1341,47 @@ var ( poolReconcileConfirmDelay = 30 * time.Second ) -// Marker files (variables so tests can point them at a temp dir). PoolJoinedAtFilePath records when HandlePoolJoin -// last wrote a pool into the config (see PoolJoinGracePeriod); PoolReconcileClearedAtFilePath when -// ReconcilePoolConfig last cleared one (see PoolReconcileMinInterval). +// Marker files (variables so tests can point them at a temp dir): +// - PoolMemberConfirmedFilePath holds ":" that this firmware has seen the Blox's cluster peer be a +// member of. ReconcilePoolConfig only ever clears a configured pool it confirmed this way — a Blox that WAS a +// member and no longer is (e.g. it left from the app while offline). A configured pool never seen as a +// membership (a join that never landed on-chain, or any config from before this firmware that isn't a +// membership) is left alone, so devices in that state keep running exactly as before. +// - PoolReconcileClearedAtFilePath: when ReconcilePoolConfig last cleared one (see PoolReconcileMinInterval). +// - PoolReconcileDisabledFilePath: while it exists, ReconcilePoolConfig does nothing (operations kill switch). var ( - PoolJoinedAtFilePath = "/internal/.tmp/pool_joined_at.tmp" + PoolMemberConfirmedFilePath = "/internal/.tmp/pool_member_confirmed.tmp" PoolReconcileClearedAtFilePath = "/internal/.tmp/pool_reconcile_cleared_at.tmp" + PoolReconcileDisabledFilePath = "/internal/.disable_pool_reconcile" ) -func recordPoolJoinTime(t time.Time) { - if err := writeTimeMarker(PoolJoinedAtFilePath, t); err != nil { - log.Warnw("Failed to record pool join time", "err", err) +func poolKey(chainName, pool string) string { return chainName + ":" + pool } + +func recordPoolMemberConfirmed(chainName, pool string) { + if confirmed, ok := readPoolMemberConfirmed(); ok && confirmed == poolKey(chainName, pool) { + return + } + if err := os.MkdirAll(filepath.Dir(PoolMemberConfirmedFilePath), 0755); err != nil { + log.Warnw("Failed to create directory for the confirmed pool membership", "err", err) + return + } + if err := os.WriteFile(PoolMemberConfirmedFilePath, []byte(poolKey(chainName, pool)), 0644); err != nil { + log.Warnw("Failed to record the confirmed pool membership", "err", err) } } -func clearPoolJoinTime() { - if err := os.Remove(PoolJoinedAtFilePath); err != nil && !os.IsNotExist(err) { - log.Warnw("Failed to remove pool join time", "err", err) +func readPoolMemberConfirmed() (string, bool) { + b, err := os.ReadFile(PoolMemberConfirmedFilePath) + if err != nil { + return "", false } + return strings.TrimSpace(string(b)), true } -// PoolJoinedAt returns when the configured pool was joined from this device; false when unknown (e.g. joins made -// before this marker existed), which counts as an old join. -func PoolJoinedAt() (time.Time, bool) { - return readTimeMarker(PoolJoinedAtFilePath) +func clearPoolMemberConfirmed() { + if err := os.Remove(PoolMemberConfirmedFilePath); err != nil && !os.IsNotExist(err) { + log.Warnw("Failed to remove the confirmed pool membership", "err", err) + } } // signalPoolConfigCleared runs the WithOnPoolConfigCleared hook (the blox wires it to a fula services restart, so @@ -1453,20 +1469,22 @@ func (bl *FxBlockchain) strictEthCall(ctx context.Context, chainName string, cal return rpcResponse.Result, nil } -// ReconcilePoolConfig clears the configured pool when the chain definitively says this device is neither a member -// of it nor waiting to join it — e.g. it was removed from the app while offline, or the app's leave notice never -// arrived — unless the pool was joined from this device less than PoolJoinGracePeriod ago. Any read error keeps the -// config. Returns true when the config was cleared (and the services restart was signalled). +// ReconcilePoolConfig clears the configured pool when this firmware has previously seen the Blox be a member of it +// (PoolMemberConfirmedFilePath) and the chain now definitively says it is neither a member nor waiting to join — i.e. +// it left (e.g. from the app while offline, or the app's leave notice never arrived). It also records memberships it +// sees, which is what makes a later leave eligible. A configured pool that was never seen as a membership is never +// touched. Any read error keeps the config. Returns true when the config was cleared (and the services restart was +// signalled). func (bl *FxBlockchain) ReconcilePoolConfig(ctx context.Context) (bool, error) { - if bl.getPoolName == nil || bl.getChainName == nil { + if _, err := os.Stat(PoolReconcileDisabledFilePath); err == nil { + log.Debugw("Pool reconcile disabled by flag file", "path", PoolReconcileDisabledFilePath) return false, nil } - pool, chainName := bl.getPoolName(), bl.getChainName() - if pool == "" || pool == "0" || chainName == "" { + if bl.poolHost || bl.getPoolName == nil || bl.getChainName == nil { return false, nil } - if joinedAt, ok := PoolJoinedAt(); ok && time.Since(joinedAt) < PoolJoinGracePeriod { - log.Debugw("Pool reconcile: join is recent, keeping config", "pool", pool, "chain", chainName, "joinedAt", joinedAt) + pool, chainName := bl.getPoolName(), bl.getChainName() + if pool == "" || pool == "0" || chainName == "" { return false, nil } // At most one reconcile clear (and services restart) per PoolReconcileMinInterval, whatever the RPCs answer. @@ -1491,9 +1509,25 @@ func (bl *FxBlockchain) ReconcilePoolConfig(ctx context.Context) (bool, error) { if err != nil { return false, err } - if member || pending { + if member { + recordPoolMemberConfirmed(chainName, pool) + return false, nil + } + if pending { return false, nil } + if check == 1 { + if confirmed, ok := readPoolMemberConfirmed(); !ok || confirmed != poolKey(chainName, pool) { + log.Debugw("Pool reconcile: configured pool was never seen as a membership of this Blox; leaving it alone", + "pool", pool, "chain", chainName) + return false, nil + } + } + } + // Abort if the config changed while checking (e.g. a join or rejoin landed meanwhile). + if bl.getPoolName() != pool || bl.getChainName() != chainName { + log.Infow("Pool reconcile: config changed while checking; leaving it alone", "pool", pool, "chain", chainName) + return false, nil } // Record the clear before doing it: if the record can't be written, the rate limit couldn't hold, so don't clear. if err := writeTimeMarker(PoolReconcileClearedAtFilePath, time.Now()); err != nil { diff --git a/blockchain/options.go b/blockchain/options.go index c4cfd26..a6173a1 100644 --- a/blockchain/options.go +++ b/blockchain/options.go @@ -37,6 +37,7 @@ type ( signingKey crypto.PrivKey // Private key for signing outgoing requests (mobile client) clientProtocolID string // Protocol ID for kubo p2p forwarding (e.g. "/x/fula-blockchain") onPoolConfigCleared func() // Called after a leave / reconcile cleared the pool from the config + poolHost bool // Pool host (--poolHost): never leaves / reconciles its pool } ) @@ -255,6 +256,14 @@ func WithOnPoolConfigCleared(fn func()) Option { } } +// WithPoolHost marks the node as a pool host (--poolHost): pool leave requests and pool reconcile never clear its pool. +func WithPoolHost(b bool) Option { + return func(o *options) error { + o.poolHost = b + return nil + } +} + // WithClientProtocolID sets the libp2p protocol ID used when dialing through kubo p2p. func WithClientProtocolID(pid string) Option { return func(o *options) error { diff --git a/blockchain/pool_leave_live_test.go b/blockchain/pool_leave_live_test.go index 6ccf1c0..c89245b 100644 --- a/blockchain/pool_leave_live_test.go +++ b/blockchain/pool_leave_live_test.go @@ -52,8 +52,9 @@ func livePeer(t *testing.T, s string) peer.ID { func newLiveBlockchain(t *testing.T, clusterPeer peer.ID, cfg *poolConfig) *FxBlockchain { t.Helper() dir := t.TempDir() - PoolJoinedAtFilePath = dir + "/pool_joined_at.tmp" + PoolMemberConfirmedFilePath = dir + "/pool_member_confirmed.tmp" PoolReconcileClearedAtFilePath = dir + "/pool_reconcile_cleared_at.tmp" + PoolReconcileDisabledFilePath = dir + "/disable_pool_reconcile" poolLeaveCheckInterval = 2 * time.Second poolConfigClearedRestartDelay = 0 poolReconcileConfirmDelay = 2 * time.Second @@ -141,11 +142,21 @@ func TestLiveReconcilePoolConfig(t *testing.T) { cleared, err := newLiveBlockchain(t, member, cfg).ReconcilePoolConfig(t.Context()) require.NoError(t, err) assert.False(t, cleared, "a real member keeps its pool") + confirmed, ok := readPoolMemberConfirmed() + assert.True(t, ok) + assert.Equal(t, chain+":1", confirmed, "and its membership is recorded") cfg = &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} cleared, err = newLiveBlockchain(t, own, cfg).ReconcilePoolConfig(t.Context()) require.NoError(t, err) - assert.True(t, cleared, "a peer that isn't in the pool has it cleared") + assert.False(t, cleared, "a pool never seen as a membership is left alone") + + cfg = &poolConfig{pool: "1", chain: chain, cleared: make(chan struct{}, 1)} + bl := newLiveBlockchain(t, own, cfg) + recordPoolMemberConfirmed(chain, "1") // it was a member once, and the chain no longer lists it + cleared, err = bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.True(t, cleared, "a former member that left has the pool cleared") waitCleared(t, cfg) } diff --git a/blockchain/pool_leave_test.go b/blockchain/pool_leave_test.go index d727206..c1b65c0 100644 --- a/blockchain/pool_leave_test.go +++ b/blockchain/pool_leave_test.go @@ -31,6 +31,8 @@ type fakeChain struct { status uint8 // joinRequests(...).status failRPC bool // answer every call with a JSON-RPC error calls int + memberCalls int + onMemberCall func(n int) // runs on each isPeerIdMemberOfPool call (1-based count) } func (f *fakeChain) RoundTrip(req *http.Request) (*http.Response, error) { @@ -56,6 +58,10 @@ func (f *fakeChain) RoundTrip(req *http.Request) (*http.Response, error) { case f.failRPC: resp["error"] = map[string]interface{}{"code": -32000, "message": "upstream unavailable"} case strings.HasPrefix(data, abi.MethodSignatures.IsPeerIdMemberOfPool): + f.memberCalls++ + if f.onMemberCall != nil { + f.onMemberCall(f.memberCalls) + } answer := f.member if len(f.memberAnswers) > 0 { answer, f.memberAnswers = f.memberAnswers[0], f.memberAnswers[1:] @@ -106,11 +112,12 @@ func (c *poolConfig) setChain(ch string) error { return nil } -func newLeaveTestBlockchain(t *testing.T, chain *fakeChain, cfg *poolConfig, withClusterPeer bool) *FxBlockchain { +func newLeaveTestBlockchain(t *testing.T, chain *fakeChain, cfg *poolConfig, withClusterPeer bool, extra ...Option) *FxBlockchain { t.Helper() dir := t.TempDir() - PoolJoinedAtFilePath = filepath.Join(dir, "pool_joined_at.tmp") + PoolMemberConfirmedFilePath = filepath.Join(dir, "pool_member_confirmed.tmp") PoolReconcileClearedAtFilePath = filepath.Join(dir, "pool_reconcile_cleared_at.tmp") + PoolReconcileDisabledFilePath = filepath.Join(dir, "disable_pool_reconcile") poolLeaveCheckInterval = 10 * time.Millisecond poolConfigClearedRestartDelay = 0 poolReconcileConfirmDelay = 0 @@ -135,7 +142,7 @@ func newLeaveTestBlockchain(t *testing.T, chain *fakeChain, cfg *poolConfig, wit require.NoError(t, err) opts = append(opts, WithClusterPeerID(id)) } - bl, err := NewFxBlockchain(NewSimpleKeyStorer(""), opts...) + bl, err := NewFxBlockchain(NewSimpleKeyStorer(""), append(opts, extra...)...) require.NoError(t, err) bl.ch = &http.Client{Transport: chain} return bl @@ -161,7 +168,7 @@ func waitCleared(t *testing.T, cfg *poolConfig) { func TestHandlePoolLeave_NotMemberClearsConfigAndRestarts(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) - recordPoolJoinTime(time.Now()) // a leave clears even a fresh join + recordPoolMemberConfirmed("skale", "1") rec := leave(bl, 1, "skale") @@ -172,8 +179,8 @@ func TestHandlePoolLeave_NotMemberClearsConfigAndRestarts(t *testing.T) { assert.Equal(t, "skale", res.ChainName) assert.Equal(t, "0", cfg.getPool()) assert.Equal(t, "", cfg.getChain()) - _, ok := PoolJoinedAt() - assert.False(t, ok, "join marker is removed on leave") + _, ok := readPoolMemberConfirmed() + assert.False(t, ok, "the confirmed membership is dropped on leave") waitCleared(t, cfg) } @@ -249,6 +256,18 @@ func TestHandlePoolLeave_NothingConfiguredIs202WithoutRestart(t *testing.T) { assert.Equal(t, 0, cfg.hookCalls, "no services restart when there was nothing to clear") } +func TestHandlePoolLeave_PoolHostNeverLeaves(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + chain := &fakeChain{} + bl := newLeaveTestBlockchain(t, chain, cfg, true, WithPoolHost(true)) + + rec := leave(bl, 1, "skale") + + assert.Equal(t, http.StatusConflict, rec.Code) + assert.Equal(t, 0, chain.calls) + assert.Equal(t, "1", cfg.getPool()) +} + func TestConfigWriteFailureNeverRestarts(t *testing.T) { // A restart after a failed write would find the pool again, clear, restart again — a loop. t.Run("leave", func(t *testing.T) { @@ -262,6 +281,7 @@ func TestConfigWriteFailureNeverRestarts(t *testing.T) { t.Run("reconcile", func(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1), failWrites: true} bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + recordPoolMemberConfirmed("skale", "1") cleared, err := bl.ReconcilePoolConfig(t.Context()) assert.Error(t, err) assert.False(t, cleared) @@ -289,62 +309,93 @@ func TestHandlePoolLeave_UnverifiableIs503AndKeepsConfig(t *testing.T) { } func TestReconcilePoolConfig(t *testing.T) { - t.Run("not a member and no join marker: cleared", func(t *testing.T) { + t.Run("running device whose pool was never seen as a membership: left alone", func(t *testing.T) { + // e.g. a join whose on-chain part never landed — such devices keep running exactly as before the update. cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) cleared, err := bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) - assert.True(t, cleared) - assert.Equal(t, "0", cfg.getPool()) - assert.Equal(t, "", cfg.getChain()) - waitCleared(t, cfg) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + assert.Equal(t, "skale", cfg.getChain()) + _, recorded := readTimeMarker(PoolReconcileClearedAtFilePath) + assert.False(t, recorded) + assert.Equal(t, 0, cfg.hookCalls) }) - t.Run("recent join: kept without reading the chain", func(t *testing.T) { - cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} - chain := &fakeChain{} - bl := newLeaveTestBlockchain(t, chain, cfg, true) - recordPoolJoinTime(time.Now().Add(-time.Hour)) + t.Run("member: kept, and the membership is recorded", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "base", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{member: true}, cfg, true) cleared, err := bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) assert.False(t, cleared) - assert.Equal(t, "1", cfg.getPool()) - assert.Equal(t, 0, chain.calls) + confirmed, ok := readPoolMemberConfirmed() + require.True(t, ok) + assert.Equal(t, "base:1", confirmed) }) - t.Run("join older than the grace period: cleared", func(t *testing.T) { - cfg := &poolConfig{pool: "1", chain: "base", cleared: make(chan struct{}, 1)} + t.Run("was a member, has left: cleared and services restarted", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) - recordPoolJoinTime(time.Now().Add(-PoolJoinGracePeriod - time.Minute)) + recordPoolMemberConfirmed("skale", "1") cleared, err := bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) assert.True(t, cleared) + assert.Equal(t, "0", cfg.getPool()) + assert.Equal(t, "", cfg.getChain()) waitCleared(t, cfg) }) - t.Run("still a member or pending: kept", func(t *testing.T) { - for _, chain := range []*fakeChain{{member: true}, {status: 1}} { + t.Run("membership confirmed for another pool or chain: left alone", func(t *testing.T) { + for _, confirmed := range [][2]string{{"base", "1"}, {"skale", "2"}} { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} - bl := newLeaveTestBlockchain(t, chain, cfg, true) + bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + recordPoolMemberConfirmed(confirmed[0], confirmed[1]) cleared, err := bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) - assert.False(t, cleared) + assert.False(t, cleared, "confirmed %v", confirmed) assert.Equal(t, "1", cfg.getPool()) } }) + t.Run("pending: kept", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, &fakeChain{status: 1}, cfg, true) + recordPoolMemberConfirmed("skale", "1") + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", cfg.getPool()) + }) t.Run("rpc error: kept, error returned", func(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} bl := newLeaveTestBlockchain(t, &fakeChain{failRPC: true}, cfg, true) + recordPoolMemberConfirmed("skale", "1") cleared, err := bl.ReconcilePoolConfig(t.Context()) assert.Error(t, err) assert.False(t, cleared) assert.Equal(t, "1", cfg.getPool()) assert.Equal(t, 0, cfg.hookCalls) }) - t.Run("no pool configured: nothing to do", func(t *testing.T) { - cfg := &poolConfig{pool: "0", chain: "", cleared: make(chan struct{}, 1)} + t.Run("nothing to do without a pool, on a pool host, or with the kill switch", func(t *testing.T) { chain := &fakeChain{} - bl := newLeaveTestBlockchain(t, chain, cfg, true) - cleared, err := bl.ReconcilePoolConfig(t.Context()) + unset := &poolConfig{pool: "0", chain: "", cleared: make(chan struct{}, 1)} + cleared, err := newLeaveTestBlockchain(t, chain, unset, true).ReconcilePoolConfig(t.Context()) require.NoError(t, err) assert.False(t, cleared) + + host := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl := newLeaveTestBlockchain(t, chain, host, true, WithPoolHost(true)) + recordPoolMemberConfirmed("skale", "1") + cleared, err = bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + + off := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + bl = newLeaveTestBlockchain(t, chain, off, true) + recordPoolMemberConfirmed("skale", "1") + require.NoError(t, os.WriteFile(PoolReconcileDisabledFilePath, nil, 0644)) + cleared, err = bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, "1", off.getPool()) + assert.Equal(t, 0, chain.calls) }) } @@ -354,14 +405,16 @@ func TestReconcilePoolConfigLoopGuards(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} chain := &fakeChain{} bl := newLeaveTestBlockchain(t, chain, cfg, true) + recordPoolMemberConfirmed("skale", "1") cleared, err := bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) require.True(t, cleared) waitCleared(t, cfg) - // Something (e.g. discovery after the restart) puts a pool back; the chain still doesn't list the Blox. + // Something puts the same pool back and it is (somehow) recorded as confirmed again; still rate-limited. require.NoError(t, cfg.setPool("1")) require.NoError(t, cfg.setChain("skale")) + recordPoolMemberConfirmed("skale", "1") calls := chain.calls cleared, err = bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) @@ -373,6 +426,7 @@ func TestReconcilePoolConfigLoopGuards(t *testing.T) { t.Run("a second read that disagrees keeps the config", func(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} bl := newLeaveTestBlockchain(t, &fakeChain{memberAnswers: []bool{false, true}}, cfg, true) + recordPoolMemberConfirmed("skale", "1") cleared, err := bl.ReconcilePoolConfig(t.Context()) require.NoError(t, err) assert.False(t, cleared) @@ -380,9 +434,25 @@ func TestReconcilePoolConfigLoopGuards(t *testing.T) { _, recorded := readTimeMarker(PoolReconcileClearedAtFilePath) assert.False(t, recorded) }) + t.Run("a join that lands while checking is not wiped", func(t *testing.T) { + cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} + chain := &fakeChain{onMemberCall: func(n int) { + if n == 2 { // between the two reads: the user joins pool 2 + _ = cfg.setPool("2") + } + }} + bl := newLeaveTestBlockchain(t, chain, cfg, true) + recordPoolMemberConfirmed("skale", "1") + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + assert.False(t, cleared) + assert.Equal(t, "2", cfg.getPool()) + assert.Equal(t, 0, cfg.hookCalls) + }) t.Run("if the clear can't be recorded, nothing is cleared", func(t *testing.T) { cfg := &poolConfig{pool: "1", chain: "skale", cleared: make(chan struct{}, 1)} bl := newLeaveTestBlockchain(t, &fakeChain{}, cfg, true) + recordPoolMemberConfirmed("skale", "1") blocker := filepath.Join(t.TempDir(), "not-a-dir") require.NoError(t, os.WriteFile(blocker, []byte("x"), 0644)) PoolReconcileClearedAtFilePath = filepath.Join(blocker, "marker.tmp") // parent is a file → unwritable @@ -395,14 +465,15 @@ func TestReconcilePoolConfigLoopGuards(t *testing.T) { }) } -func TestHandlePoolJoinRecordsJoinTime(t *testing.T) { - PoolJoinedAtFilePath = filepath.Join(t.TempDir(), "pool_joined_at.tmp") - before := time.Now().Add(-time.Second) - recordPoolJoinTime(time.Now()) - at, ok := PoolJoinedAt() +func TestPoolMemberConfirmedMarker(t *testing.T) { + PoolMemberConfirmedFilePath = filepath.Join(t.TempDir(), "pool_member_confirmed.tmp") + _, ok := readPoolMemberConfirmed() + assert.False(t, ok) + recordPoolMemberConfirmed("base", "3") + got, ok := readPoolMemberConfirmed() require.True(t, ok) - assert.True(t, at.After(before)) - clearPoolJoinTime() - _, err := os.Stat(PoolJoinedAtFilePath) + assert.Equal(t, "base:3", got) + clearPoolMemberConfirmed() + _, err := os.Stat(PoolMemberConfirmedFilePath) assert.True(t, os.IsNotExist(err)) } diff --git a/blox/blox.go b/blox/blox.go index 288464c..581d059 100644 --- a/blox/blox.go +++ b/blox/blox.go @@ -96,6 +96,7 @@ func New(o ...Option) (*Blox, error) { blockchain.WithIpfsClient(p.rpc), blockchain.WithMinSuccessPingCount(p.minSuccessRate*p.pingCount/100), blockchain.WithIpfsClusterAPI(p.ipfsClusterApi), + blockchain.WithPoolHost(p.poolHostMode), // After a pool leave / reconcile cleared the config, restart the fula services so ipfs-cluster re-reads // the (now empty) pool instead of following the old one until the next reboot. blockchain.WithOnPoolConfigCleared(func() { @@ -434,11 +435,6 @@ func (p *Blox) Start(ctx context.Context) error { p.watchKuboP2P(ctx) }() - // Drop a configured pool this Blox no longer belongs to on-chain (left from the app while it was offline, or the - // leave notice never arrived). Runs in the background a little after start, never on pool hosts; a clear - // restarts the fula services, after which discovery runs as for a fresh device. - p.startPoolReconcile(ctx) - // Register cluster tunnel forward immediately if pool is already known from config, // before potentially blocking on chain discovery for up to ~15 minutes. if p.topicName != "0" { @@ -577,6 +573,11 @@ func (p *Blox) Start(ctx context.Context) error { } } + // Only now (pool discovery above is done, so the two never touch the config at the same time): record this + // Blox's pool membership, and drop a pool it was a member of but has since left (e.g. from the app while it was + // offline). Background, a little later, never on pool hosts; a clear restarts the fula services. + p.startPoolReconcile(ctx) + if err := p.bl.FetchUsersAndPopulateSets(ctx, p.topicName, true, 15*time.Second); err != nil { log.Errorw("FetchUsersAndPopulateSets failed", "err", err) } diff --git a/cmd/blox/main.go b/cmd/blox/main.go index 841bfa2..13a665b 100644 --- a/cmd/blox/main.go +++ b/cmd/blox/main.go @@ -542,7 +542,14 @@ func before(ctx *cli.Context) error { return os.WriteFile(app.configPath, yc, 0700) } +// configMu serialises the runtime config.yaml accessors below: they all read the file into the shared app.config +// and may write it back, and they are called from several goroutines (join / leave handlers, pool discovery, +// pool reconcile, the 6 h loop). +var configMu sync.Mutex + func getPoolNameFromConfig() string { + configMu.Lock() + defer configMu.Unlock() configData, err := os.ReadFile(app.configPath) if err != nil { return "0" @@ -556,6 +563,8 @@ func getPoolNameFromConfig() string { } func getChainNameFromConfig() string { + configMu.Lock() + defer configMu.Unlock() configData, err := os.ReadFile(app.configPath) if err != nil { return "" @@ -569,6 +578,8 @@ func getChainNameFromConfig() string { } func updateChainName(newChainName string) error { + configMu.Lock() + defer configMu.Unlock() // Load existing config file configData, err := os.ReadFile(app.configPath) if err != nil { @@ -602,6 +613,8 @@ func updateChainName(newChainName string) error { } func updateConfig(p []peer.ID) error { + configMu.Lock() + defer configMu.Unlock() // Load existing config file configData, err := os.ReadFile(app.configPath) if err != nil { @@ -648,6 +661,8 @@ func updateConfig(p []peer.ID) error { } func updatePoolName(newPoolName string) error { + configMu.Lock() + defer configMu.Unlock() // Load existing config file configData, err := os.ReadFile(app.configPath) if err != nil { From 6a09169bfd827f1ef118843156b5c15b7103120a Mon Sep 17 00:00:00 2001 From: ehsan shariati Date: Tue, 6 Oct 2026 01:32:56 -0400 Subject: [PATCH 6/6] docs(blockchain): note why the confirmed-membership marker needs no lock Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QaKz9HeXaNBjrchbbGimFx --- blockchain/bl_pool.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index 58fdc50..699f3c4 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -1357,6 +1357,8 @@ var ( func poolKey(chainName, pool string) string { return chainName + ":" + pool } +// recordPoolMemberConfirmed is not locked: its only caller, ReconcilePoolConfig, runs once a few minutes after start +// and then every 6 h, so calls never overlap (the leave handler only deletes the file). Add a lock if that changes. func recordPoolMemberConfirmed(chainName, pool string) { if confirmed, ok := readPoolMemberConfirmed(); ok && confirmed == poolKey(chainName, pool) { return