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..699f3c4 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" @@ -324,6 +326,9 @@ func (bl *FxBlockchain) cleanLeaveJoinPool(ctx context.Context, PoolID int) { log.Errorw("Failed to reset chain name", "error", err) } } + // (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.) + clearPoolMemberConfirmed() bl.StopPingServer(ctx) // Announcements no longer used - pool joins handled via blockchain API @@ -1180,7 +1185,17 @@ 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; +// - 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) var req PoolLeaveRequest @@ -1197,17 +1212,47 @@ 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 + 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 { + 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 + } + // 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 @@ -1222,31 +1267,47 @@ 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()), - } - w.WriteHeader(http.StatusBadRequest) - json.NewEncoder(w).Encode(errMsg) + // 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 } - // 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()), + // 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.StatusInternalServerError) - json.NewEncoder(w).Encode(errMsg) + stillListed = member || pending + if !stillListed || attempt == poolLeaveCheckAttempts { + break + } + 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 } // 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{ @@ -1261,32 +1322,125 @@ 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 + // 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 +) + +// 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 ( + PoolMemberConfirmedFilePath = "/internal/.tmp/pool_member_confirmed.tmp" + PoolReconcileClearedAtFilePath = "/internal/.tmp/pool_reconcile_cleared_at.tmp" + PoolReconcileDisabledFilePath = "/internal/.disable_pool_reconcile" +) + +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 } + 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 readPoolMemberConfirmed() (string, bool) { + b, err := os.ReadFile(PoolMemberConfirmedFilePath) + if err != nil { + return "", false + } + return strings.TrimSpace(string(b)), true +} + +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 +// 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 +1448,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 +1462,125 @@ 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 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 _, err := os.Stat(PoolReconcileDisabledFilePath); err == nil { + log.Debugw("Pool reconcile disabled by flag file", "path", PoolReconcileDisabledFilePath) + return false, nil + } + if bl.poolHost || bl.getPoolName == nil || bl.getChainName == nil { + return false, nil + } + 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. + 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) + } + // 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 { + 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 { + 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 { + return false, err + } + bl.signalPoolConfigCleared(0) + return true, nil +} - log.Infow("Successfully called removeMemberPeerId", "poolID", poolID, "peerID", peerID, "chain", chainName) +// 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 { + 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) + } + } 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/options.go b/blockchain/options.go index 1fd3790..a6173a1 100644 --- a/blockchain/options.go +++ b/blockchain/options.go @@ -36,6 +36,8 @@ 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 + poolHost bool // Pool host (--poolHost): never leaves / reconciles its pool } ) @@ -245,6 +247,23 @@ 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 + } +} + +// 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 new file mode 100644 index 0000000..c89245b --- /dev/null +++ b/blockchain/pool_leave_live_test.go @@ -0,0 +1,165 @@ +//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() + dir := t.TempDir() + 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 + 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") + 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.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) +} + +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..c1b65c0 --- /dev/null +++ b/blockchain/pool_leave_test.go @@ -0,0 +1,479 @@ +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 + 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 + memberCalls int + onMemberCall func(n int) // runs on each isPeerIdMemberOfPool call (1-based count) +} + +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): + 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:] + } + isMember := uint64(0) + if answer { + 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 + 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 } +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() + if c.failWrites { + return fmt.Errorf("read-only file system") + } + 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, extra ...Option) *FxBlockchain { + t.Helper() + dir := t.TempDir() + 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 + + 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(""), append(opts, extra...)...) + 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) + recordPoolMemberConfirmed("skale", "1") + + 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()) + _, ok := readPoolMemberConfirmed() + assert.False(t, ok, "the confirmed membership is dropped 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_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_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) { + 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) + recordPoolMemberConfirmed("skale", "1") + 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)} + 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("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.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("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) + confirmed, ok := readPoolMemberConfirmed() + require.True(t, ok) + assert.Equal(t, "base:1", confirmed) + }) + 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) + 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("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, &fakeChain{}, cfg, true) + recordPoolMemberConfirmed(confirmed[0], confirmed[1]) + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + 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("nothing to do without a pool, on a pool host, or with the kill switch", func(t *testing.T) { + chain := &fakeChain{} + 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) + }) +} + +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) + recordPoolMemberConfirmed("skale", "1") + cleared, err := bl.ReconcilePoolConfig(t.Context()) + require.NoError(t, err) + require.True(t, cleared) + waitCleared(t, cfg) + + // 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) + 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) + recordPoolMemberConfirmed("skale", "1") + 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("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 + 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 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.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 119c5b5..581d059 100644 --- a/blox/blox.go +++ b/blox/blox.go @@ -96,6 +96,14 @@ 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() { + 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 @@ -565,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) } @@ -637,6 +650,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..1ee083c --- /dev/null +++ b/blox/pool_reconcile.go @@ -0,0 +1,47 @@ +package blox + +import ( + "context" + "time" +) + +const ( + // 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 +) + +// reconcilePool clears the configured pool when the chain definitively says this Blox left it (see +// 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() + cleared, err := p.bl.ReconcilePoolConfig(ctx) + if err != nil { + log.Warnw("Pool reconcile skipped; keeping the configured pool", "err", err) + return + } + if cleared { + 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) + }() +} 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 {