diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml deleted file mode 100644 index 2630b955..00000000 --- a/.github/workflows/build.yml +++ /dev/null @@ -1,19 +0,0 @@ -name: Build - -on: [ push, pull_request ] - -jobs: - build: - name: Container - runs-on: ubuntu-latest - steps: - - name: Checkout - uses: actions/checkout@v3 - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v2 - - name: Build Container image - uses: docker/build-push-action@v3 - with: - context: . - push: false - platforms: linux/amd64,linux/arm64 \ No newline at end of file diff --git a/.github/workflows/go-check.yml b/.github/workflows/go-check.yml index ed1dffae..f83eeeab 100644 --- a/.github/workflows/go-check.yml +++ b/.github/workflows/go-check.yml @@ -12,8 +12,11 @@ jobs: - uses: actions/setup-go@v5 with: go-version: "1.25.x" + # Analyzer tools must build with the Go above (1.25: cmd/blox doesn't build on 1.26 yet). staticcheck + # v0.7.0 (2026.1) and x/tools v0.49.0 are the last releases that do; newer ones need Go 1.26, and + # v0.4.6 no longer builds on 1.25. Bump these together with go-version. - name: Install staticcheck - run: go install honnef.co/go/tools/cmd/staticcheck@v0.4.6 + run: go install honnef.co/go/tools/cmd/staticcheck@v0.7.0 - name: Clear Go module cache run: go clean -modcache - name: Check that go.mod is tidy @@ -37,7 +40,7 @@ jobs: run: go vet ./... - name: Install shadow analyzer if: success() || failure() - run: go install golang.org/x/tools/go/analysis/passes/shadow/cmd/shadow@latest + run: go install golang.org/x/tools/go/analysis/passes/shadow/cmd/shadow@v0.49.0 # A shadowed `err` in wap/pkg/server once left the WAP API refusing to start after a SUCCESSFUL bind: # the retry loop redeclared err, so the outer err still held the original failure and Serve() logged # "Successfully bound ..." immediately followed by "... Server will not start." `go vet` does not catch diff --git a/.github/workflows/go-test.yml b/.github/workflows/go-test.yml index 55b13889..18c7d375 100644 --- a/.github/workflows/go-test.yml +++ b/.github/workflows/go-test.yml @@ -30,7 +30,9 @@ jobs: env: SECRETS_DIR: ${{ github.workspace }}/secrets + # mobile's client tests spend ~5 min waiting out the client's fixed dial timeouts (there is no device to + # reach); under -race that package alone comes close to go test's default 10 min limit. - name: Run tests with race detector - run: go test -v -race ./... + run: go test -v -race -timeout 20m ./... env: SECRETS_DIR: ${{ github.workspace }}/secrets diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml deleted file mode 100644 index d2c6b90d..00000000 --- a/.github/workflows/publish.yml +++ /dev/null @@ -1,42 +0,0 @@ -name: Publish - -on: - release: - types: - - published - -env: - REGISTRY: ghcr.io - IMAGE_NAME: ${{ github.repository }} - -jobs: - build-and-push-image: - name: Container - runs-on: ubuntu-latest - permissions: - contents: read - packages: write - steps: - - name: Checkout repository - uses: actions/checkout@v3 - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v2 - - name: Log in to the Container registry - uses: docker/login-action@v2 - with: - registry: ${{ env.REGISTRY }} - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - name: Extract metadata (tags, labels) for Docker - id: meta - uses: docker/metadata-action@v4 - with: - images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} - - name: Build and push Docker image - uses: docker/build-push-action@v3 - with: - context: . - push: true - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - platforms: linux/amd64,linux/arm64 \ No newline at end of file diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index a307be98..00000000 --- a/Dockerfile +++ /dev/null @@ -1,46 +0,0 @@ -# Build stage -FROM golang:1.23 AS buildstage - -WORKDIR /go-fula - -# Copy go module files and download dependencies -COPY ./go-fula/go.mod ./go-fula/go.sum ./ -RUN go mod download -x - -# Copy the rest of the application source code -COPY ./go-fula/ . - -# Build binaries with CGO disabled for static binaries -RUN CGO_ENABLED=0 GOOS=linux go build -o /app ./cmd/blox && \ - CGO_ENABLED=0 GOOS=linux go build -o /wap ./wap/cmd && \ - CGO_ENABLED=0 GOOS=linux go build -o /initipfs ./modules/initipfs && \ - CGO_ENABLED=0 GOOS=linux go build -o /initipfscluster ./modules/initipfscluster - -# Final stage -FROM alpine:3.20 - -# Install packages from stable repos (NM 1.46.6 — compatible with host NM 1.46.0) -# IMPORTANT: Do NOT use edge/testing for networkmanager packages — a newer nmcli -# than the host's NM daemon causes "unknown property" errors (e.g. mac-address-denylist) -RUN apk update && \ - apk add --no-cache hostapd iw wireless-tools networkmanager-wifi networkmanager-cli \ - networkmanager-dnsmasq dhcpcd iptables curl && \ - apk add --no-cache --repository http://dl-cdn.alpinelinux.org/alpine/edge/testing mergerfs - -WORKDIR / - -# Copy binaries from the build stage -COPY --from=buildstage /app /app -COPY --from=buildstage /wap /wap -COPY --from=buildstage /initipfs /initipfs -COPY --from=buildstage /initipfscluster /initipfscluster - -# Copy and set permissions for the startup script -COPY ./go-fula.sh /go-fula.sh -RUN chmod +x /go-fula.sh - -# Expose necessary ports -EXPOSE 40001 5001 - -# Set the entrypoint command -CMD ["/go-fula.sh"] \ No newline at end of file diff --git a/blockchain/abi/contract.go b/blockchain/abi/contract.go index bbd4d340..54c07b18 100644 --- a/blockchain/abi/contract.go +++ b/blockchain/abi/contract.go @@ -96,9 +96,7 @@ var MethodSignatures = struct { // DecodePoolsResult decodes the result from pools(uint32) contract call func DecodePoolsResult(data string) (*Pool, error) { // Remove 0x prefix - if strings.HasPrefix(data, "0x") { - data = data[2:] - } + data = strings.TrimPrefix(data, "0x") // Check if we have enough data (minimum 9 * 32 bytes = 288 hex chars) if len(data) < 288 { @@ -247,9 +245,7 @@ func DecodePoolsResult(data string) (*Pool, error) { // DecodeIsMemberOfPoolResult decodes the result from isPeerIdMemberOfPool(uint32,bytes32) contract call func DecodeIsMemberOfPoolResult(data string) (*IsMemberOfPoolResult, error) { // Remove 0x prefix - if strings.HasPrefix(data, "0x") { - data = data[2:] - } + data = strings.TrimPrefix(data, "0x") // Check if we have enough data (2 * 32 bytes = 128 hex chars) if len(data) < 128 { @@ -280,9 +276,7 @@ func DecodeIsMemberOfPoolResult(data string) (*IsMemberOfPoolResult, error) { // DecodeGetMemberPeerIdsResult decodes the result from getMemberPeerIds(uint32,address) contract call func DecodeGetMemberPeerIdsResult(data string) (*GetMemberPeerIdsResult, error) { // Remove 0x prefix - if strings.HasPrefix(data, "0x") { - data = data[2:] - } + data = strings.TrimPrefix(data, "0x") // Check if we have enough data (minimum 64 hex chars for offset + length) if len(data) < 64 { @@ -346,30 +340,26 @@ func EncodePoolIdsCall(index uint32) string { // DecodePoolIdResponse decodes the response from poolIds(uint256) call func DecodePoolIdResponse(result string) (uint32, error) { // Remove 0x prefix - if strings.HasPrefix(result, "0x") { - result = result[2:] - } - + result = strings.TrimPrefix(result, "0x") + // Should be 64 hex characters (32 bytes) if len(result) != 64 { return 0, fmt.Errorf("invalid response length: expected 64 hex chars, got %d", len(result)) } - + // Parse as uint64 first, then convert to uint32 poolID, err := strconv.ParseUint(result, 16, 64) if err != nil { return 0, fmt.Errorf("failed to parse pool ID: %w", err) } - + return uint32(poolID), nil } // EncodeIsPeerIdMemberOfPoolCall encodes the isPeerIdMemberOfPool(uint32,bytes32) method call func EncodeIsPeerIdMemberOfPoolCall(poolID uint32, peerIDBytes32 string) string { // Remove 0x prefix from peerIDBytes32 if present - if strings.HasPrefix(peerIDBytes32, "0x") { - peerIDBytes32 = peerIDBytes32[2:] - } + peerIDBytes32 = strings.TrimPrefix(peerIDBytes32, "0x") return fmt.Sprintf("%s%064x%s", MethodSignatures.IsPeerIdMemberOfPool, poolID, peerIDBytes32) } @@ -377,9 +367,7 @@ func EncodeIsPeerIdMemberOfPoolCall(poolID uint32, peerIDBytes32 string) string // EncodeGetMemberPeerIdsCall encodes the getMemberPeerIds(uint32,address) method call func EncodeGetMemberPeerIdsCall(poolID uint32, memberAddress string) string { // Remove 0x prefix from memberAddress if present - if strings.HasPrefix(memberAddress, "0x") { - memberAddress = memberAddress[2:] - } + memberAddress = strings.TrimPrefix(memberAddress, "0x") // Pad address to 32 bytes (64 hex chars) paddedAddress := fmt.Sprintf("%064s", memberAddress) @@ -390,9 +378,7 @@ func EncodeGetMemberPeerIdsCall(poolID uint32, memberAddress string) string { // EncodeRemoveMemberPeerIdCall encodes the removeMemberPeerId(uint32,bytes32) method call func EncodeRemoveMemberPeerIdCall(poolID uint32, peerIDBytes32 string) string { // Remove 0x prefix from peerIDBytes32 if present - if strings.HasPrefix(peerIDBytes32, "0x") { - peerIDBytes32 = peerIDBytes32[2:] - } + peerIDBytes32 = strings.TrimPrefix(peerIDBytes32, "0x") return fmt.Sprintf("%s%064x%s", MethodSignatures.RemoveMemberPeerId, poolID, peerIDBytes32) } @@ -400,9 +386,7 @@ func EncodeRemoveMemberPeerIdCall(poolID uint32, peerIDBytes32 string) string { // 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:] - } + peerIDBytes32 = strings.TrimPrefix(peerIDBytes32, "0x") return fmt.Sprintf("%s%064x%s", MethodSignatures.JoinRequests, poolID, peerIDBytes32) } @@ -412,9 +396,7 @@ func EncodeJoinRequestsCall(poolID uint32, peerIDBytes32 string) string { // 0 = no request. func DecodeJoinRequestStatus(data string) (uint8, error) { // Remove 0x prefix - if strings.HasPrefix(data, "0x") { - data = data[2:] - } + data = strings.TrimPrefix(data, "0x") // 8 * 32 bytes = 512 hex chars if len(data) < 512 { diff --git a/blockchain/bl_account.go b/blockchain/bl_account.go index 1e4c1390..9153661c 100644 --- a/blockchain/bl_account.go +++ b/blockchain/bl_account.go @@ -16,7 +16,7 @@ import ( func (bl *FxBlockchain) Seeded(ctx context.Context, to peer.ID, r SeededRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -46,7 +46,7 @@ func (bl *FxBlockchain) Seeded(ctx context.Context, to peer.ID, r SeededRequest) func (bl *FxBlockchain) AccountCreate(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionAccountCreate, nil) @@ -72,7 +72,7 @@ func (bl *FxBlockchain) AccountCreate(ctx context.Context, to peer.ID) ([]byte, func (bl *FxBlockchain) AccountFund(ctx context.Context, to peer.ID, r AccountFundRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -104,7 +104,7 @@ func (bl *FxBlockchain) AccountFund(ctx context.Context, to peer.ID, r AccountFu func (bl *FxBlockchain) AccountExists(ctx context.Context, to peer.ID, r AccountExistsRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -135,7 +135,7 @@ func (bl *FxBlockchain) AccountExists(ctx context.Context, to peer.ID, r Account func (bl *FxBlockchain) AccountBalance(ctx context.Context, to peer.ID, r AccountBalanceRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -165,7 +165,7 @@ func (bl *FxBlockchain) AccountBalance(ctx context.Context, to peer.ID, r Accoun func (bl *FxBlockchain) AssetsBalance(ctx context.Context, to peer.ID, r AssetsBalanceRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -195,7 +195,7 @@ func (bl *FxBlockchain) AssetsBalance(ctx context.Context, to peer.ID, r AssetsB func (bl *FxBlockchain) TransferToFula(ctx context.Context, to peer.ID, r TransferToFulaRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer diff --git a/blockchain/bl_autopin.go b/blockchain/bl_autopin.go index 53ad7b2b..178dede1 100644 --- a/blockchain/bl_autopin.go +++ b/blockchain/bl_autopin.go @@ -171,7 +171,7 @@ func (bl *FxBlockchain) handleAutoPinUnpair(from peer.ID, w http.ResponseWriter, // AutoPinPair is the P2P client-side method for the mobile bridge. func (bl *FxBlockchain) AutoPinPair(ctx context.Context, to peer.ID, r AutoPinPairRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -203,7 +203,7 @@ func (bl *FxBlockchain) AutoPinPair(ctx context.Context, to peer.ID, r AutoPinPa // AutoPinRefresh is the P2P client-side method for the mobile bridge. func (bl *FxBlockchain) AutoPinRefresh(ctx context.Context, to peer.ID, r AutoPinRefreshRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -235,7 +235,7 @@ func (bl *FxBlockchain) AutoPinRefresh(ctx context.Context, to peer.ID, r AutoPi // AutoPinUnpair is the P2P client-side method for the mobile bridge. func (bl *FxBlockchain) AutoPinUnpair(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://"+to.String()+".invalid/"+actionAutoPinUnpair, nil) diff --git a/blockchain/bl_manifest.go b/blockchain/bl_manifest.go index 7f9f21fd..05981917 100644 --- a/blockchain/bl_manifest.go +++ b/blockchain/bl_manifest.go @@ -20,7 +20,7 @@ import ( func (bl *FxBlockchain) ManifestUpload(ctx context.Context, to peer.ID, r ManifestUploadRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -57,7 +57,7 @@ func (bl *FxBlockchain) ManifestUpload(ctx context.Context, to peer.ID, r Manife func (bl *FxBlockchain) ManifestStore(ctx context.Context, to peer.ID, r ManifestStoreRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -234,7 +234,7 @@ func (bl *FxBlockchain) HandleManifestsAvailable(ctx context.Context, poolIDStri func (bl *FxBlockchain) ManifestAvailable(ctx context.Context, to peer.ID, r ManifestAvailableRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -265,7 +265,7 @@ func (bl *FxBlockchain) ManifestAvailable(ctx context.Context, to peer.ID, r Man func (bl *FxBlockchain) ManifestBatchStore(ctx context.Context, to peer.ID, r ManifestBatchStoreRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -296,7 +296,7 @@ func (bl *FxBlockchain) ManifestBatchStore(ctx context.Context, to peer.ID, r Ma func (bl *FxBlockchain) ManifestBatchUpload(ctx context.Context, to peer.ID, r ManifestBatchUploadMobileRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -327,7 +327,7 @@ func (bl *FxBlockchain) ManifestBatchUpload(ctx context.Context, to peer.ID, r M func (bl *FxBlockchain) ManifestRemove(ctx context.Context, to peer.ID, r ManifestRemoveRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -364,7 +364,7 @@ func (bl *FxBlockchain) ManifestRemove(ctx context.Context, to peer.ID, r Manife func (bl *FxBlockchain) ManifestRemoveStorer(ctx context.Context, to peer.ID, r ManifestRemoveStorerRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -401,7 +401,7 @@ func (bl *FxBlockchain) ManifestRemoveStorer(ctx context.Context, to peer.ID, r func (bl *FxBlockchain) ManifestRemoveStored(ctx context.Context, to peer.ID, r ManifestRemoveStoredRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer diff --git a/blockchain/bl_nas.go b/blockchain/bl_nas.go index bf0dcbec..7c2c6112 100644 --- a/blockchain/bl_nas.go +++ b/blockchain/bl_nas.go @@ -110,7 +110,7 @@ func (bl *FxBlockchain) handleNasCredentials(from peer.ID, w http.ResponseWriter // NasCredentials is the P2P client-side method for retrieving the NAS credentials. func (bl *FxBlockchain) NasCredentials(ctx context.Context, to peer.ID, r NasCredentialsRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer diff --git a/blockchain/bl_plugins.go b/blockchain/bl_plugins.go index 719c8de9..b71adef7 100644 --- a/blockchain/bl_plugins.go +++ b/blockchain/bl_plugins.go @@ -283,7 +283,7 @@ func (bl *FxBlockchain) uninstallPluginImpl(ctx context.Context, pluginName stri func (bl *FxBlockchain) ListPlugins(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionListPlugins, nil) @@ -300,7 +300,7 @@ func (bl *FxBlockchain) ListPlugins(ctx context.Context, to peer.ID) ([]byte, er func (bl *FxBlockchain) InstallPlugin(ctx context.Context, to peer.ID, pluginName string, paramsString string) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -322,7 +322,7 @@ func (bl *FxBlockchain) InstallPlugin(ctx context.Context, to peer.ID, pluginNam func (bl *FxBlockchain) UninstallPlugin(ctx context.Context, to peer.ID, pluginName string) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -344,7 +344,7 @@ func (bl *FxBlockchain) UninstallPlugin(ctx context.Context, to peer.ID, pluginN func (bl *FxBlockchain) ListActivePlugins(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionListActivePlugins, nil) @@ -483,7 +483,7 @@ func (bl *FxBlockchain) showPluginStatusImpl(ctx context.Context, pluginName str func (bl *FxBlockchain) GetInstallOutput(ctx context.Context, to peer.ID, pluginName string, paramsString string) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -505,7 +505,7 @@ func (bl *FxBlockchain) GetInstallOutput(ctx context.Context, to peer.ID, plugin func (bl *FxBlockchain) GetInstallStatus(ctx context.Context, to peer.ID, pluginName string) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -723,7 +723,7 @@ func (bl *FxBlockchain) HandleUpdatePlugin(w http.ResponseWriter, r *http.Reques func (bl *FxBlockchain) UpdatePlugin(ctx context.Context, to peer.ID, pluginName string) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer diff --git a/blockchain/bl_pool.go b/blockchain/bl_pool.go index 699f3c44..79292686 100644 --- a/blockchain/bl_pool.go +++ b/blockchain/bl_pool.go @@ -28,7 +28,7 @@ type PingResponse struct { func (bl *FxBlockchain) PoolCreate(ctx context.Context, to peer.ID, r PoolCreateRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -149,7 +149,7 @@ func (bl *FxBlockchain) HandlePoolJoin(method string, action string, from peer.I func (bl *FxBlockchain) PoolJoin(ctx context.Context, to peer.ID, r PoolJoinRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -194,7 +194,7 @@ func (bl *FxBlockchain) StopPingServer(ctx context.Context) error { func (bl *FxBlockchain) PoolCancelJoin(ctx context.Context, to peer.ID, r PoolCancelJoinRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -344,7 +344,7 @@ func (bl *FxBlockchain) cleanLeaveJoinPool(ctx context.Context, PoolID int) { func (bl *FxBlockchain) PoolRequests(ctx context.Context, to peer.ID, r PoolRequestsRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -381,7 +381,7 @@ func (bl *FxBlockchain) PoolRequests(ctx context.Context, to peer.ID, r PoolRequ func (bl *FxBlockchain) PoolList(ctx context.Context, to peer.ID, r PoolListRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -418,7 +418,7 @@ func (bl *FxBlockchain) PoolList(ctx context.Context, to peer.ID, r PoolListRequ func (bl *FxBlockchain) ReplicateInPool(ctx context.Context, to peer.ID, r ReplicateRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -455,7 +455,7 @@ func (bl *FxBlockchain) ReplicateInPool(ctx context.Context, to peer.ID, r Repli func (bl *FxBlockchain) PoolUserList(ctx context.Context, to peer.ID, r PoolUserListRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -492,7 +492,7 @@ func (bl *FxBlockchain) PoolUserList(ctx context.Context, to peer.ID, r PoolUser func (bl *FxBlockchain) PoolVote(ctx context.Context, to peer.ID, r PoolVoteRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -529,7 +529,7 @@ func (bl *FxBlockchain) PoolVote(ctx context.Context, to peer.ID, r PoolVoteRequ func (bl *FxBlockchain) PoolLeave(ctx context.Context, to peer.ID, r PoolLeaveRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -713,7 +713,7 @@ func (bl *FxBlockchain) HandleEVMPoolList(ctx context.Context, chainName string) ctx, cancel := context.WithTimeout(ctx, time.Second*time.Duration(bl.timeout)) defer cancel() - chainConfigs := GetChainConfigs() + chainConfigs := bl.chainConfigs() chainConfig, exists := chainConfigs[chainName] if !exists { return EVMPoolListResponse{}, fmt.Errorf("unsupported chain: %s", chainName) @@ -881,7 +881,7 @@ func (bl *FxBlockchain) HandleIsMemberOfPool(ctx context.Context, req IsMemberOf ctx, cancel := context.WithTimeout(ctx, time.Second*time.Duration(bl.timeout)) defer cancel() - chainConfigs := GetChainConfigs() + chainConfigs := bl.chainConfigs() chainConfig, exists := chainConfigs[req.ChainName] if !exists { return IsMemberOfPoolResponse{}, fmt.Errorf("unsupported chain: %s", req.ChainName) @@ -976,7 +976,7 @@ func (bl *FxBlockchain) GetPoolCreatorPeerID(ctx context.Context, poolID uint32, ctx, cancel := context.WithTimeout(ctx, time.Second*time.Duration(bl.timeout)) defer cancel() - chainConfigs := GetChainConfigs() + chainConfigs := bl.chainConfigs() chainConfig, exists := chainConfigs[chainName] if !exists { return "", fmt.Errorf("unsupported chain: %s", chainName) @@ -1129,7 +1129,7 @@ func (bl *FxBlockchain) discoverPoolChain(ctx context.Context, poolID uint32) (s // validatePoolOnChain checks if a pool exists on a specific chain func (bl *FxBlockchain) validatePoolOnChain(ctx context.Context, poolID uint32, chainName string) error { - chainConfigs := GetChainConfigs() + chainConfigs := bl.chainConfigs() chainConfig, exists := chainConfigs[chainName] if !exists { return fmt.Errorf("unsupported chain: %s", chainName) @@ -1437,7 +1437,7 @@ func (bl *FxBlockchain) ClusterPeerPoolStatus(ctx context.Context, poolID uint32 // 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] + chainConfig, exists := bl.chainConfigs()[chainName] if !exists { return "", fmt.Errorf("unsupported chain: %s", chainName) } diff --git a/blockchain/blockchain.go b/blockchain/blockchain.go index 78ebea86..b806f4c4 100644 --- a/blockchain/blockchain.go +++ b/blockchain/blockchain.go @@ -67,8 +67,6 @@ type ( fetchCheckStop chan struct{} stopFetchUsersAfterJoinChan chan struct{} - cachedAccount string - isAccountCached bool fetchMutex sync.Mutex isFetching bool @@ -284,9 +282,18 @@ func GetChainConfigs() map[string]ChainConfig { } } +// chainConfigs returns the chain configurations this instance talks to: GetChainConfigs(), unless a test replaced +// them with withChainConfigs. +func (bl *FxBlockchain) chainConfigs() map[string]ChainConfig { + if bl.chainConfigOverride != nil { + return bl.chainConfigOverride + } + return GetChainConfigs() +} + // callEVMChain makes calls to EVM-compatible chains (Base/Skale) using JSON-RPC func (bl *FxBlockchain) callEVMChain(ctx context.Context, chainName string, method string, params []interface{}) ([]byte, int, error) { - chainConfigs := GetChainConfigs() + chainConfigs := bl.chainConfigs() chainConfig, exists := chainConfigs[chainName] if !exists { return nil, http.StatusBadRequest, fmt.Errorf("unsupported chain: %s", chainName) @@ -1060,11 +1067,6 @@ func contains(slice []string, str string) bool { return false } -func (bl *FxBlockchain) cleanUnwantedPeers(keepPeers []peer.ID) { - // No-op: go-fula no longer has a libp2p host or peerstore. - // Peer management is handled by kubo. -} - func (bl *FxBlockchain) checkIfUserHasOpenPoolRequests(ctx context.Context, topicString string) (bool, error) { topic, err := strconv.Atoi(topicString) if err != nil { @@ -1331,77 +1333,6 @@ func saveCreatorPeerID(poolID int, peerID string) error { return os.WriteFile(filename, []byte(peerID), 0644) // Adjust permissions if needed } -func fetchPoolDetails(ctx context.Context, bl *FxBlockchain, poolID int) (*Pool, error) { - req := PoolListRequestWithPoolId{PoolID: poolID} - action := "actionPoolList" - - responseBody, statusCode, err := bl.callBlockchain(ctx, "POST", action, req) - if err != nil { - return nil, fmt.Errorf("blockchain call error: %w, status code: %d", err, statusCode) - } - - if statusCode != http.StatusOK { - var errMsg map[string]interface{} - if jsonErr := json.Unmarshal(responseBody, &errMsg); jsonErr == nil { - return nil, fmt.Errorf("unexpected response status: %d, message: %s, description: %s", - statusCode, errMsg["message"], errMsg["description"]) - } else { - return nil, fmt.Errorf("unexpected response status: %d, body: %s", statusCode, string(responseBody)) - } - } - - var response PoolListResponse - if err := json.Unmarshal(responseBody, &response); err != nil { - return nil, err - } - - for _, pool := range response.Pools { - if pool.PoolID == poolID { - return &pool, nil - } - } - - return nil, fmt.Errorf("pool with ID %d not found", poolID) -} - -func (bl *FxBlockchain) fetchUserDetails(ctx context.Context, poolID int) (*PoolUserListResponse, error) { - req := PoolUserListRequest{ - PoolID: poolID, - } - action := "actionPoolUserList" - - responseBody, statusCode, err := bl.callBlockchain(ctx, "POST", action, req) - if err != nil { - return nil, fmt.Errorf("blockchain call error: %w, status code: %d", err, statusCode) - } - - if statusCode != http.StatusOK { - var errMsg map[string]interface{} - if jsonErr := json.Unmarshal(responseBody, &errMsg); jsonErr == nil { - return nil, fmt.Errorf("unexpected response status: %d, message: %s, description: %s", - statusCode, errMsg["message"], errMsg["description"]) - } else { - return nil, fmt.Errorf("unexpected response status: %d, body: %s", statusCode, string(responseBody)) - } - } - - var response PoolUserListResponse - if err := json.Unmarshal(responseBody, &response); err != nil { - return nil, err - } - - return &response, nil -} - -func (bl *FxBlockchain) findPeerID(creatorClusterPeerID string, userDetails *PoolUserListResponse) string { - for _, user := range userDetails.Users { - if user.Account == creatorClusterPeerID { - return user.PeerID - } - } - return "" -} - func (bl *FxBlockchain) handleActionManifestBatchUpload(method string, action string, from peer.ID, w http.ResponseWriter, r *http.Request, req *ManifestBatchUploadRequest) { log := log.With("action", action, "from", from) res := reflect.New(responseTypes[action]).Interface() diff --git a/blockchain/blockchain_test.go b/blockchain/blockchain_test.go index e9331210..b2812a36 100644 --- a/blockchain/blockchain_test.go +++ b/blockchain/blockchain_test.go @@ -58,15 +58,36 @@ func TestFxBlockchainCreation(t *testing.T) { WithAllowTransientConnection(true), WithBlockchainEndPoint("127.0.0.1:4000"), WithTimeout(30), + withListenAddrs("127.0.0.1:0", "127.0.0.1:0"), ) require.NoError(t, err) require.NotNil(t, bl) + t.Cleanup(func() { _ = bl.Shutdown(context.Background()) }) // Test that blockchain can start err = bl.Start(ctx) assert.NoError(t, err) } +// TestDefaultListenAddrsAndChainConfigs pins what the daemon gets when no test hook is set. +func TestDefaultListenAddrsAndChainConfigs(t *testing.T) { + bl, err := NewFxBlockchain(NewSimpleKeyStorer(""), WithGetPoolName(nil), WithGetChainName(nil), + WithUpdatePoolName(nil), WithUpdateChainName(nil)) + require.NoError(t, err) + t.Cleanup(func() { _ = bl.Shutdown(context.Background()) }) + + assert.Equal(t, "0.0.0.0:4020", bl.proxyListenAddr) + assert.Equal(t, "0.0.0.0:4021", bl.pingListenAddr) + assert.Equal(t, GetChainConfigs(), bl.chainConfigs()) + // nil getters/setters keep the defaults instead of leaving nil funcs to panic on. + require.NotNil(t, bl.getPoolName) + require.NotNil(t, bl.getChainName) + require.NotNil(t, bl.updatePoolName) + require.NotNil(t, bl.updateChainName) + assert.Equal(t, "0", bl.getPoolName()) + assert.Equal(t, "", bl.getChainName()) +} + // TestCallEVMChain tests EVM chain calls with mock server func TestCallEVMChain(t *testing.T) { // Create mock EVM RPC server @@ -181,7 +202,7 @@ func TestCallEVMChainWithRetry(t *testing.T) { // This test demonstrates the intended behavior // The actual implementation would need to be modified to support dependency injection - _, _, err = bl.callEVMChainWithRetry(ctx, "test", "eth_call", params, 3) + _, _, _ = bl.callEVMChainWithRetry(ctx, "test", "eth_call", params, 3) // For now, we just verify the method exists assert.NotNil(t, bl) @@ -220,8 +241,10 @@ func TestBlockchainHealthCheck(t *testing.T) { WithSelfPeerID(h.ID()), WithAuthorizer(h.ID()), WithTimeout(30), + withListenAddrs("127.0.0.1:0", "127.0.0.1:0"), ) require.NoError(t, err) + t.Cleanup(func() { _ = bl.Shutdown(context.Background()) }) ctx := context.Background() diff --git a/blockchain/blox.go b/blockchain/blox.go index 042b0b85..db1a81d7 100644 --- a/blockchain/blox.go +++ b/blockchain/blox.go @@ -70,7 +70,7 @@ func (b *StreamBuffer) IsClosed() bool { func (bl *FxBlockchain) BloxFreeSpace(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionBloxFreeSpace, nil) @@ -97,7 +97,7 @@ func (bl *FxBlockchain) BloxFreeSpace(ctx context.Context, to peer.ID) ([]byte, func (bl *FxBlockchain) EraseBlData(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionEraseBlData, nil) @@ -124,7 +124,7 @@ func (bl *FxBlockchain) EraseBlData(ctx context.Context, to peer.ID) ([]byte, er func (bl *FxBlockchain) WifiRemoveall(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionWifiRemoveall, nil) @@ -151,7 +151,7 @@ func (bl *FxBlockchain) WifiRemoveall(ctx context.Context, to peer.ID) ([]byte, func (bl *FxBlockchain) Reboot(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionReboot, nil) @@ -178,7 +178,7 @@ func (bl *FxBlockchain) Reboot(ctx context.Context, to peer.ID) ([]byte, error) func (bl *FxBlockchain) Partition(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionPartition, nil) @@ -205,7 +205,7 @@ func (bl *FxBlockchain) Partition(ctx context.Context, to peer.ID) ([]byte, erro func (bl *FxBlockchain) DeleteFulaConfig(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionDeleteFulaConfig, nil) @@ -231,7 +231,7 @@ func (bl *FxBlockchain) DeleteFulaConfig(ctx context.Context, to peer.ID) ([]byt } func (bl *FxBlockchain) GetAccount(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionGetAccount, nil) @@ -259,7 +259,7 @@ func (bl *FxBlockchain) GetAccount(ctx context.Context, to peer.ID) ([]byte, err func (bl *FxBlockchain) FetchContainerLogs(ctx context.Context, to peer.ID, r wifi.FetchContainerLogsRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -295,7 +295,7 @@ func (bl *FxBlockchain) FetchContainerLogs(ctx context.Context, to peer.ID, r wi func (bl *FxBlockchain) ChatWithAI(ctx context.Context, to peer.ID, r wifi.ChatWithAIRequest) (*StreamBuffer, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -375,7 +375,7 @@ func (bl *FxBlockchain) ChatWithAI(ctx context.Context, to peer.ID, r wifi.ChatW func (bl *FxBlockchain) FindBestAndTargetInLogs(ctx context.Context, to peer.ID, r wifi.FindBestAndTargetInLogsRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -412,7 +412,7 @@ func (bl *FxBlockchain) FindBestAndTargetInLogs(ctx context.Context, to peer.ID, func (bl *FxBlockchain) GetFolderSize(ctx context.Context, to peer.ID, r wifi.GetFolderSizeRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -449,7 +449,7 @@ func (bl *FxBlockchain) GetFolderSize(ctx context.Context, to peer.ID, r wifi.Ge func (bl *FxBlockchain) GetDatastoreSize(ctx context.Context, to peer.ID, r wifi.GetDatastoreSizeRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -485,7 +485,7 @@ func (bl *FxBlockchain) GetDatastoreSize(ctx context.Context, to peer.ID, r wifi func (bl *FxBlockchain) GetDockerImageBuildDates(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionGetDockerImageBuildDates, nil) @@ -511,7 +511,7 @@ func (bl *FxBlockchain) GetDockerImageBuildDates(ctx context.Context, to peer.ID func (bl *FxBlockchain) GetClusterInfo(ctx context.Context, to peer.ID) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+to.String()+".invalid/"+actionGetClusterInfo, nil) @@ -538,7 +538,7 @@ func (bl *FxBlockchain) GetClusterInfo(ctx context.Context, to peer.ID) ([]byte, func (bl *FxBlockchain) DeleteWifi(ctx context.Context, to peer.ID, r wifi.DeleteWifiRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -575,7 +575,7 @@ func (bl *FxBlockchain) DeleteWifi(ctx context.Context, to peer.ID, r wifi.Delet func (bl *FxBlockchain) DisconnectWifi(ctx context.Context, to peer.ID, r wifi.DeleteWifiRequest) ([]byte, error) { if bl.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.blockchain") + ctx = network.WithAllowLimitedConn(ctx, "fx.blockchain") } var buf bytes.Buffer @@ -621,10 +621,7 @@ func (bl *FxBlockchain) handleFetchContainerLogs(ctx context.Context, from peer. } log.Debugw("handleFetchContainerLogs received", "req", req) - out := wifi.FetchContainerLogsResponse{ - Status: true, - Msg: "", - } + var out wifi.FetchContainerLogsResponse res, err := wifi.FetchContainerLogs(ctx, req) if err != nil { out = wifi.FetchContainerLogsResponse{ @@ -751,11 +748,7 @@ func (bl *FxBlockchain) handleFindBestAndTargetInLogs(ctx context.Context, from } log.Debugw("handleFindBestAndTargetInLogs received", "req", req) - out := wifi.FindBestAndTargetInLogsResponse{ - Best: "0", - Target: "0", - Err: "", - } + var out wifi.FindBestAndTargetInLogsResponse best, target, err := wifi.FindBestAndTargetInLogs(ctx, req) if err != nil { out = wifi.FindBestAndTargetInLogsResponse{ @@ -792,10 +785,7 @@ func (bl *FxBlockchain) handleGetFolderSize(ctx context.Context, from peer.ID, w } log.Debugw("handleGetFolderSize received", "req", req) - out := wifi.GetFolderSizeResponse{ - FolderPath: "", - SizeInBytes: "", - } + var out wifi.GetFolderSizeResponse res, err := wifi.GetFolderSize(ctx, req) if err != nil { out = wifi.GetFolderSizeResponse{ @@ -827,7 +817,7 @@ func (bl *FxBlockchain) handleGetDatastoreSize(ctx context.Context, from peer.ID } log.Debugw("handleGetDatastoreSize received", "req", req) - out := wifi.GetDatastoreSizeResponse{} + var out wifi.GetDatastoreSizeResponse res, err := wifi.GetDatastoreSize(ctx, req) if err != nil { out = wifi.GetDatastoreSizeResponse{} diff --git a/blockchain/integration_test.go b/blockchain/integration_test.go index 86345fda..9dfb89b0 100644 --- a/blockchain/integration_test.go +++ b/blockchain/integration_test.go @@ -1,9 +1,15 @@ +//go:build livechain + +// Calls the real SKALE and Base pool contracts, so it needs network and follows live chain state: run it with +// `go test -tags livechain ./blockchain/`. + package blockchain import ( "context" "crypto/rand" "encoding/json" + "strings" "testing" "time" @@ -157,8 +163,9 @@ func TestRealContractIntegration(t *testing.T) { t.Logf(" Members: %d/%d", poolData.MemberCount, poolData.MaxMembers) // Verify the decoded data matches expected values + expectedName := map[string]string{"skale": "Global-S1", "base": "Global-B1"}[chainName] assert.Equal(t, uint32(1), poolData.ID, "Pool ID should be 1") - assert.Equal(t, "Global-S1", poolData.Name, "Pool name should be Global-S1") + assert.Equal(t, expectedName, poolData.Name, "Pool 1 name on %s", chainName) assert.Equal(t, "global", poolData.Region, "Pool region should be global") }) } @@ -249,7 +256,9 @@ func TestRealContractIntegration(t *testing.T) { // - Base: false, 0x0000000000000000000000000000000000000000 if chainName == "skale" { assert.True(t, membershipResult.IsMember, "Peer should be member on Skale") - assert.Equal(t, "0xCe12f8cE914dA115191De28f2E1796a24E475B72", membershipResult.MemberAddress, "Member address should match on Skale") + // The decoder returns lowercase hex, so compare without the EIP-55 checksum casing. + assert.True(t, strings.EqualFold("0xCe12f8cE914dA115191De28f2E1796a24E475B72", membershipResult.MemberAddress), + "Member address should match on Skale, got %s", membershipResult.MemberAddress) } else if chainName == "base" { assert.False(t, membershipResult.IsMember, "Peer should not be member on Base") assert.Equal(t, "0x0000000000000000000000000000000000000000", membershipResult.MemberAddress, "Member address should be zero on Base") diff --git a/blockchain/interface.go b/blockchain/interface.go index ceb4f7e7..f84453e4 100644 --- a/blockchain/interface.go +++ b/blockchain/interface.go @@ -38,21 +38,21 @@ const ( actionManifestAvailableAllaccountsBatch = "fula-manifest-available_allaccounts_batch" //Hardware - actionBloxFreeSpace = "blox-free-space" - actionEraseBlData = "erase-blockchain-data" - actionWifiRemoveall = "wifi-removeall" - actionReboot = "reboot" - actionPartition = "partition" - actionDeleteFulaConfig = "delete-fula-config" - actionDeleteWifi = "delete-wifi" - actionDisconnectWifi = "disconnect-wifi" - actionGetAccount = "get-account" - actionFetchContainerLogs = "fetch-container-logs" - actionFindBestAndTargetInLogs = "find-bestandtarget-inlogs" - actionGetFolderSize = "get-folder-size" - actionGetDatastoreSize = "get-datastore-size" - actionGetDockerImageBuildDates = "get-docker-image-build-dates" - actionGetClusterInfo = "get-cluster-info" + actionBloxFreeSpace = "blox-free-space" + actionEraseBlData = "erase-blockchain-data" + actionWifiRemoveall = "wifi-removeall" + actionReboot = "reboot" + actionPartition = "partition" + actionDeleteFulaConfig = "delete-fula-config" + actionDeleteWifi = "delete-wifi" + actionDisconnectWifi = "disconnect-wifi" + actionGetAccount = "get-account" + actionFetchContainerLogs = "fetch-container-logs" + actionFindBestAndTargetInLogs = "find-bestandtarget-inlogs" + actionGetFolderSize = "get-folder-size" + actionGetDatastoreSize = "get-datastore-size" + actionGetDockerImageBuildDates = "get-docker-image-build-dates" + actionGetClusterInfo = "get-cluster-info" // Cluster actionReplicateInPool = "replicate" @@ -656,21 +656,21 @@ var requestTypes = map[string]reflect.Type{ actionReplicateInPool: reflect.TypeOf(ReplicateRequest{}), //Hardware - actionBloxFreeSpace: reflect.TypeOf(wifi.BloxFreeSpaceRequest{}), - actionEraseBlData: reflect.TypeOf(wifi.EraseBlDataRequest{}), - actionWifiRemoveall: reflect.TypeOf(wifi.WifiRemoveallRequest{}), - actionReboot: reflect.TypeOf(wifi.RebootRequest{}), - actionPartition: reflect.TypeOf(wifi.PartitionRequest{}), - actionDeleteFulaConfig: reflect.TypeOf(wifi.DeleteFulaConfigRequest{}), - actionDeleteWifi: reflect.TypeOf(wifi.DeleteWifiRequest{}), - actionDisconnectWifi: reflect.TypeOf(wifi.DeleteWifiRequest{}), - actionGetAccount: reflect.TypeOf(GetAccountRequest{}), - actionFetchContainerLogs: reflect.TypeOf(wifi.FetchContainerLogsRequest{}), - actionFindBestAndTargetInLogs: reflect.TypeOf(wifi.FindBestAndTargetInLogsRequest{}), - actionGetFolderSize: reflect.TypeOf(wifi.GetFolderSizeRequest{}), - actionGetDatastoreSize: reflect.TypeOf(wifi.GetDatastoreSizeRequest{}), - actionGetDockerImageBuildDates: reflect.TypeOf(wifi.GetDockerImageBuildDatesRequest{}), - actionGetClusterInfo: reflect.TypeOf(wifi.GetClusterInfoRequest{}), + actionBloxFreeSpace: reflect.TypeOf(wifi.BloxFreeSpaceRequest{}), + actionEraseBlData: reflect.TypeOf(wifi.EraseBlDataRequest{}), + actionWifiRemoveall: reflect.TypeOf(wifi.WifiRemoveallRequest{}), + actionReboot: reflect.TypeOf(wifi.RebootRequest{}), + actionPartition: reflect.TypeOf(wifi.PartitionRequest{}), + actionDeleteFulaConfig: reflect.TypeOf(wifi.DeleteFulaConfigRequest{}), + actionDeleteWifi: reflect.TypeOf(wifi.DeleteWifiRequest{}), + actionDisconnectWifi: reflect.TypeOf(wifi.DeleteWifiRequest{}), + actionGetAccount: reflect.TypeOf(GetAccountRequest{}), + actionFetchContainerLogs: reflect.TypeOf(wifi.FetchContainerLogsRequest{}), + actionFindBestAndTargetInLogs: reflect.TypeOf(wifi.FindBestAndTargetInLogsRequest{}), + actionGetFolderSize: reflect.TypeOf(wifi.GetFolderSizeRequest{}), + actionGetDatastoreSize: reflect.TypeOf(wifi.GetDatastoreSizeRequest{}), + actionGetDockerImageBuildDates: reflect.TypeOf(wifi.GetDockerImageBuildDatesRequest{}), + actionGetClusterInfo: reflect.TypeOf(wifi.GetClusterInfoRequest{}), // Plugins actionListPlugins: reflect.TypeOf(ListPluginsRequest{}), @@ -723,21 +723,21 @@ var responseTypes = map[string]reflect.Type{ actionReplicateInPool: reflect.TypeOf(ReplicateResponse{}), //Hardware - actionBloxFreeSpace: reflect.TypeOf(wifi.BloxFreeSpaceResponse{}), - actionEraseBlData: reflect.TypeOf(wifi.EraseBlDataResponse{}), - actionWifiRemoveall: reflect.TypeOf(wifi.WifiRemoveallResponse{}), - actionReboot: reflect.TypeOf(wifi.RebootResponse{}), - actionPartition: reflect.TypeOf(wifi.PartitionResponse{}), - actionDeleteFulaConfig: reflect.TypeOf(wifi.DeleteFulaConfigResponse{}), - actionDeleteWifi: reflect.TypeOf(wifi.DeleteWifiResponse{}), - actionDisconnectWifi: reflect.TypeOf(wifi.DeleteWifiResponse{}), - actionGetAccount: reflect.TypeOf(GetAccountResponse{}), - actionFetchContainerLogs: reflect.TypeOf(wifi.FetchContainerLogsResponse{}), - actionFindBestAndTargetInLogs: reflect.TypeOf(wifi.FindBestAndTargetInLogsResponse{}), - actionGetFolderSize: reflect.TypeOf(wifi.GetFolderSizeResponse{}), - actionGetDatastoreSize: reflect.TypeOf(wifi.GetDatastoreSizeResponse{}), - actionGetDockerImageBuildDates: reflect.TypeOf(wifi.GetDockerImageBuildDatesResponse{}), - actionGetClusterInfo: reflect.TypeOf(wifi.GetClusterInfoResponse{}), + actionBloxFreeSpace: reflect.TypeOf(wifi.BloxFreeSpaceResponse{}), + actionEraseBlData: reflect.TypeOf(wifi.EraseBlDataResponse{}), + actionWifiRemoveall: reflect.TypeOf(wifi.WifiRemoveallResponse{}), + actionReboot: reflect.TypeOf(wifi.RebootResponse{}), + actionPartition: reflect.TypeOf(wifi.PartitionResponse{}), + actionDeleteFulaConfig: reflect.TypeOf(wifi.DeleteFulaConfigResponse{}), + actionDeleteWifi: reflect.TypeOf(wifi.DeleteWifiResponse{}), + actionDisconnectWifi: reflect.TypeOf(wifi.DeleteWifiResponse{}), + actionGetAccount: reflect.TypeOf(GetAccountResponse{}), + actionFetchContainerLogs: reflect.TypeOf(wifi.FetchContainerLogsResponse{}), + actionFindBestAndTargetInLogs: reflect.TypeOf(wifi.FindBestAndTargetInLogsResponse{}), + actionGetFolderSize: reflect.TypeOf(wifi.GetFolderSizeResponse{}), + actionGetDatastoreSize: reflect.TypeOf(wifi.GetDatastoreSizeResponse{}), + actionGetDockerImageBuildDates: reflect.TypeOf(wifi.GetDockerImageBuildDatesResponse{}), + actionGetClusterInfo: reflect.TypeOf(wifi.GetClusterInfoResponse{}), // Plugins actionListPlugins: reflect.TypeOf(ListPluginsResponse{}), diff --git a/blockchain/options.go b/blockchain/options.go index a6173a11..a4cb698f 100644 --- a/blockchain/options.go +++ b/blockchain/options.go @@ -32,12 +32,15 @@ type ( fetchFrequency time.Duration //Hours that it should update the list of pool users and pool requests if not called through pubsub rpc *rpc.HttpApi ipfsClusterApi ipfsCluster.Client - selfPeerID peer.ID // Peer ID derived from private key, used for authorization checks - 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 + selfPeerID peer.ID // Peer ID derived from private key, used for authorization checks + 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 + proxyListenAddr string // TCP address of the kubo-forwarded blockchain proxy + pingListenAddr string // TCP address of the kubo-forwarded ping server + chainConfigOverride map[string]ChainConfig // Replaces GetChainConfigs() when set (tests only) } ) @@ -74,6 +77,8 @@ func newOptions(o ...Option) (*options, error) { fetchFrequency: time.Hour * 1, // default frequency, e.g., 1 hour rpc: nil, ipfsClusterApi: nil, + proxyListenAddr: ProxyListenAddr, + pingListenAddr: PingListenAddr, } for _, apply := range o { if err := apply(&opts); err != nil { @@ -177,16 +182,23 @@ func WithTopicName(n string) Option { } } +// WithUpdatePoolName sets the pool name setter. A nil setter keeps the default (callers such as blox.New pass +// their own option through even when it was never set). func WithUpdatePoolName(updatePoolName func(string) error) Option { return func(o *options) error { - o.updatePoolName = updatePoolName + if updatePoolName != nil { + o.updatePoolName = updatePoolName + } return nil } } +// WithGetPoolName sets the pool name getter. A nil getter keeps the default. func WithGetPoolName(getPoolName func() string) Option { return func(o *options) error { - o.getPoolName = getPoolName + if getPoolName != nil { + o.getPoolName = getPoolName + } return nil } } @@ -198,16 +210,22 @@ func WithChainName(n string) Option { } } +// WithUpdateChainName sets the chain name setter. A nil setter keeps the default. func WithUpdateChainName(updateChainName func(string) error) Option { return func(o *options) error { - o.updateChainName = updateChainName + if updateChainName != nil { + o.updateChainName = updateChainName + } return nil } } +// WithGetChainName sets the chain name getter. A nil getter keeps the default. func WithGetChainName(getChainName func() string) Option { return func(o *options) error { - o.getChainName = getChainName + if getChainName != nil { + o.getChainName = getChainName + } return nil } } @@ -271,3 +289,26 @@ func WithClientProtocolID(pid string) Option { return nil } } + +// withListenAddrs moves the proxy and ping servers off ProxyListenAddr / PingListenAddr (tests use +// "127.0.0.1:0" so parallel or repeated Starts don't collide). Empty keeps the default. +func withListenAddrs(proxy, ping string) Option { + return func(o *options) error { + if proxy != "" { + o.proxyListenAddr = proxy + } + if ping != "" { + o.pingListenAddr = ping + } + return nil + } +} + +// withChainConfigs replaces the built-in chain configurations (RPC endpoints, contracts), so tests can point the +// EVM calls at a mock server. +func withChainConfigs(c map[string]ChainConfig) Option { + return func(o *options) error { + o.chainConfigOverride = c + return nil + } +} diff --git a/blockchain/peerid_utils.go b/blockchain/peerid_utils.go index 520134f5..b4d9a070 100644 --- a/blockchain/peerid_utils.go +++ b/blockchain/peerid_utils.go @@ -62,9 +62,7 @@ func peerIdToBytes32(peerID string) (string, error) { // Remove the 'z' prefix for comparison originalWithoutZ := peerID - if strings.HasPrefix(originalWithoutZ, "z") { - originalWithoutZ = originalWithoutZ[1:] - } + originalWithoutZ = strings.TrimPrefix(originalWithoutZ, "z") if reconstructed != originalWithoutZ { return "", fmt.Errorf("could not revert the encoded bytes32 back to original PeerID. Got: %s, Expected: %s", reconstructed, originalWithoutZ) @@ -77,9 +75,7 @@ func peerIdToBytes32(peerID string) (string, error) { // Always returns a multibase-style PeerID (without the 'z' prefix by default) func bytes32ToPeerId(digestBytes32 string) (string, error) { // Remove 0x prefix if present - if strings.HasPrefix(digestBytes32, "0x") { - digestBytes32 = digestBytes32[2:] - } + digestBytes32 = strings.TrimPrefix(digestBytes32, "0x") // Decode hex string to bytes pubkeyBytes, err := hex.DecodeString(digestBytes32) diff --git a/blockchain/ping_server.go b/blockchain/ping_server.go index 341ce917..f5b3d77d 100644 --- a/blockchain/ping_server.go +++ b/blockchain/ping_server.go @@ -34,7 +34,7 @@ type PingServerResponse struct { // Any HTTP request to the server returns a JSON response with basic node info, // including ipfs-cluster health when the cluster is reachable. func (bl *FxBlockchain) StartPingProxy(ctx context.Context) error { - listener, err := net.Listen("tcp", PingListenAddr) + listener, err := net.Listen("tcp", bl.pingListenAddr) if err != nil { return err } @@ -88,7 +88,7 @@ func (bl *FxBlockchain) StartPingProxy(ctx context.Context) error { log.Errorw("Ping server stopped erroneously", "err", err) } }() - log.Infow("Ping proxy server started", "addr", PingListenAddr) + log.Infow("Ping proxy server started", "addr", bl.pingListenAddr) return nil } diff --git a/blockchain/pool_bug_test.go b/blockchain/pool_bug_test.go index 2aa6c283..794dbeae 100644 --- a/blockchain/pool_bug_test.go +++ b/blockchain/pool_bug_test.go @@ -1,3 +1,7 @@ +//go:build livechain + +// Calls the real SKALE and Base pool contracts: run with `go test -tags livechain ./blockchain/`. + package blockchain import ( @@ -10,7 +14,7 @@ import ( // TestPoolDiscoveryBug - Simple test to isolate the pool discovery issue func TestPoolDiscoveryBug(t *testing.T) { fmt.Println("=== TESTING POOL DISCOVERY BUG ===") - + // Create blockchain instance with proper initialization bl, err := NewFxBlockchain( NewSimpleKeyStorer(""), @@ -31,12 +35,12 @@ func TestPoolDiscoveryBug(t *testing.T) { } fmt.Printf("Found %d pools\n", len(poolList.Pools)) - + // Print each discovered pool for i, pool := range poolList.Pools { - fmt.Printf("Pool[%d]: ID=%d, Name='%s', Creator='%s', ChainName='%s'\n", + fmt.Printf("Pool[%d]: ID=%d, Name='%s', Creator='%s', ChainName='%s'\n", i, pool.ID, pool.Name, pool.Creator, pool.ChainName) - + // Verify pool ID is not 0 if pool.ID == 0 { t.Errorf("ERROR: Pool at index %d has ID 0, which should not exist!", i) @@ -45,10 +49,10 @@ func TestPoolDiscoveryBug(t *testing.T) { // Now test membership check for each discovered pool testPeerID := "12D3KooWGjK8GLeFxYQmthm5rNmbvbiA3S4zbYorzAA63RhKaYc1" - + for _, pool := range poolList.Pools { fmt.Printf("\n--- Testing membership for pool ID %d ---\n", pool.ID) - + req := IsMemberOfPoolRequest{ PeerID: testPeerID, PoolID: pool.ID, @@ -56,29 +60,29 @@ func TestPoolDiscoveryBug(t *testing.T) { } fmt.Printf("Calling HandleIsMemberOfPool with PoolID=%d\n", req.PoolID) - + resp, err := bl.HandleIsMemberOfPool(ctx, req) if err != nil { t.Errorf("Membership check failed for pool %d: %v", pool.ID, err) continue } - fmt.Printf("Membership result: IsMember=%t, MemberAddress=%s, PoolID=%d, ChainName=%s\n", + fmt.Printf("Membership result: IsMember=%t, MemberAddress=%s, PoolID=%d, ChainName=%s\n", resp.IsMember, resp.MemberAddress, resp.PoolID, resp.ChainName) - + // Check if the response has the correct pool ID if resp.PoolID != pool.ID { t.Errorf("ERROR: Response pool ID mismatch! Expected %d, got %d", pool.ID, resp.PoolID) } } - + fmt.Println("\n=== TEST COMPLETE ===") } // TestDirectMembershipCheck - Test membership check directly with known pool ID func TestDirectMembershipCheck(t *testing.T) { fmt.Println("=== TESTING DIRECT MEMBERSHIP CHECK ===") - + bl, err := NewFxBlockchain( NewSimpleKeyStorer(""), WithTimeout(30), @@ -94,7 +98,7 @@ func TestDirectMembershipCheck(t *testing.T) { // Test pool 1 directly (we know it exists) fmt.Println("Testing membership for pool ID 1 directly...") - + req := IsMemberOfPoolRequest{ PeerID: testPeerID, PoolID: 1, // Directly test pool 1 @@ -102,18 +106,18 @@ func TestDirectMembershipCheck(t *testing.T) { } fmt.Printf("Calling HandleIsMemberOfPool with PoolID=%d\n", req.PoolID) - + resp, err := bl.HandleIsMemberOfPool(ctx, req) if err != nil { t.Fatalf("Direct membership check failed for pool 1: %v", err) } - fmt.Printf("Direct membership result: IsMember=%t, MemberAddress=%s, PoolID=%d, ChainName=%s\n", + fmt.Printf("Direct membership result: IsMember=%t, MemberAddress=%s, PoolID=%d, ChainName=%s\n", resp.IsMember, resp.MemberAddress, resp.PoolID, resp.ChainName) // Test pool 0 directly (should return false immediately) fmt.Println("\nTesting membership for pool ID 0 directly...") - + req0 := IsMemberOfPoolRequest{ PeerID: testPeerID, PoolID: 0, // Test pool 0 @@ -121,18 +125,18 @@ func TestDirectMembershipCheck(t *testing.T) { } fmt.Printf("Calling HandleIsMemberOfPool with PoolID=%d\n", req0.PoolID) - + resp0, err := bl.HandleIsMemberOfPool(ctx, req0) if err != nil { t.Fatalf("Pool 0 membership check should not fail: %v", err) } - fmt.Printf("Pool 0 membership result: IsMember=%t, MemberAddress=%s, PoolID=%d, ChainName=%s\n", + fmt.Printf("Pool 0 membership result: IsMember=%t, MemberAddress=%s, PoolID=%d, ChainName=%s\n", resp0.IsMember, resp0.MemberAddress, resp0.PoolID, resp0.ChainName) if resp0.IsMember { t.Error("ERROR: Pool 0 should not have any members!") } - + fmt.Println("\n=== DIRECT TEST COMPLETE ===") } diff --git a/blockchain/pool_discovery_live_test.go b/blockchain/pool_discovery_live_test.go new file mode 100644 index 00000000..7e17b9f4 --- /dev/null +++ b/blockchain/pool_discovery_live_test.go @@ -0,0 +1,146 @@ +//go:build livechain + +// Calls the real SKALE and Base pool contracts: run with `go test -tags livechain ./blockchain/`. + +package blockchain + +import ( + "context" + "testing" + "time" +) + +// TestPoolDiscoveryAndMembership tests the complete pool discovery and membership flow +func TestPoolDiscoveryAndMembership(t *testing.T) { + // Create blockchain instance with proper options + bl, err := NewFxBlockchain( + NewSimpleKeyStorer(""), + WithTimeout(30), + ) + if err != nil { + t.Fatalf("Failed to create blockchain: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + testPeerID := "12D3KooWGjK8GLeFxYQmthm5rNmbvbiA3S4zbYorzAA63RhKaYc1" + + // Test both chains + chains := []string{"skale", "base"} + + for _, chainName := range chains { + t.Run("Chain_"+chainName, func(t *testing.T) { + t.Logf("Testing pool discovery and membership for chain: %s", chainName) + + // Step 1: Test pool discovery + poolList, err := bl.HandleEVMPoolList(ctx, chainName) + if err != nil { + t.Fatalf("Pool discovery failed for chain %s: %v", chainName, err) + } + + t.Logf("Found %d pools on chain %s", len(poolList.Pools), chainName) + + // Verify we found at least one pool + if len(poolList.Pools) == 0 { + t.Fatalf("No pools discovered on chain %s", chainName) + } + + // Verify pool IDs are not 0 + for i, pool := range poolList.Pools { + if pool.ID == 0 { + t.Errorf("Pool at index %d has ID 0, which should not exist", i) + } + t.Logf("Discovered pool ID: %d on chain %s", pool.ID, chainName) + } + + // Step 2: Test membership check for each discovered pool + for _, pool := range poolList.Pools { + t.Run("Pool_"+string(rune(pool.ID)), func(t *testing.T) { + req := IsMemberOfPoolRequest{ + PeerID: testPeerID, + PoolID: pool.ID, + ChainName: chainName, + } + + t.Logf("Checking membership for pool %d on chain %s", pool.ID, chainName) + + resp, err := bl.HandleIsMemberOfPool(ctx, req) + if err != nil { + t.Errorf("Membership check failed for pool %d on chain %s: %v", pool.ID, chainName, err) + return + } + + t.Logf("Membership result for pool %d on chain %s: isMember=%t, address=%s", + pool.ID, chainName, resp.IsMember, resp.MemberAddress) + + // Verify response structure + if resp.ChainName != chainName { + t.Errorf("Expected chain name %s, got %s", chainName, resp.ChainName) + } + if resp.PoolID != pool.ID { + t.Errorf("Expected pool ID %d, got %d", pool.ID, resp.PoolID) + } + + // For skale chain and pool 1, we expect membership to be true + if chainName == "skale" && pool.ID == 1 { + if !resp.IsMember { + t.Errorf("Expected membership to be true for pool 1 on skale chain") + } + if resp.MemberAddress == "0x0000000000000000000000000000000000000000" { + t.Errorf("Expected non-zero member address for pool 1 on skale chain") + } + } + }) + } + }) + } +} + +// TestPoolDiscoveryReturnsCorrectPoolIDs tests that pool discovery returns the correct pool IDs +func TestPoolDiscoveryReturnsCorrectPoolIDs(t *testing.T) { + bl, err := NewFxBlockchain( + NewSimpleKeyStorer(""), + WithTimeout(30), + ) + if err != nil { + t.Fatalf("Failed to create blockchain: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + chains := []string{"skale", "base"} + + for _, chainName := range chains { + t.Run("PoolIDs_"+chainName, func(t *testing.T) { + poolList, err := bl.HandleEVMPoolList(ctx, chainName) + if err != nil { + t.Fatalf("Pool discovery failed: %v", err) + } + + t.Logf("Chain %s discovered pools:", chainName) + for i, pool := range poolList.Pools { + t.Logf(" Pool[%d]: ID=%d, Name=%s, Creator=%s", i, pool.ID, pool.Name, pool.Creator) + + // Verify pool ID is valid (not 0) + if pool.ID == 0 { + t.Errorf("Pool at index %d has invalid ID 0", i) + } + } + + // We expect to find pool 1 on both chains + foundPool1 := false + for _, pool := range poolList.Pools { + if pool.ID == 1 { + foundPool1 = true + break + } + } + + if !foundPool1 { + t.Errorf("Expected to find pool ID 1 on chain %s", chainName) + } + }) + } +} diff --git a/blockchain/pool_discovery_test.go b/blockchain/pool_discovery_test.go index 573a1cea..b59c0909 100644 --- a/blockchain/pool_discovery_test.go +++ b/blockchain/pool_discovery_test.go @@ -6,93 +6,6 @@ import ( "time" ) -// TestPoolDiscoveryAndMembership tests the complete pool discovery and membership flow -func TestPoolDiscoveryAndMembership(t *testing.T) { - // Create blockchain instance with proper options - bl, err := NewFxBlockchain( - NewSimpleKeyStorer(""), - WithTimeout(30), - ) - if err != nil { - t.Fatalf("Failed to create blockchain: %v", err) - } - - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - - testPeerID := "12D3KooWGjK8GLeFxYQmthm5rNmbvbiA3S4zbYorzAA63RhKaYc1" - - // Test both chains - chains := []string{"skale", "base"} - - for _, chainName := range chains { - t.Run("Chain_"+chainName, func(t *testing.T) { - t.Logf("Testing pool discovery and membership for chain: %s", chainName) - - // Step 1: Test pool discovery - poolList, err := bl.HandleEVMPoolList(ctx, chainName) - if err != nil { - t.Fatalf("Pool discovery failed for chain %s: %v", chainName, err) - } - - t.Logf("Found %d pools on chain %s", len(poolList.Pools), chainName) - - // Verify we found at least one pool - if len(poolList.Pools) == 0 { - t.Fatalf("No pools discovered on chain %s", chainName) - } - - // Verify pool IDs are not 0 - for i, pool := range poolList.Pools { - if pool.ID == 0 { - t.Errorf("Pool at index %d has ID 0, which should not exist", i) - } - t.Logf("Discovered pool ID: %d on chain %s", pool.ID, chainName) - } - - // Step 2: Test membership check for each discovered pool - for _, pool := range poolList.Pools { - t.Run("Pool_"+string(rune(pool.ID)), func(t *testing.T) { - req := IsMemberOfPoolRequest{ - PeerID: testPeerID, - PoolID: pool.ID, - ChainName: chainName, - } - - t.Logf("Checking membership for pool %d on chain %s", pool.ID, chainName) - - resp, err := bl.HandleIsMemberOfPool(ctx, req) - if err != nil { - t.Errorf("Membership check failed for pool %d on chain %s: %v", pool.ID, chainName, err) - return - } - - t.Logf("Membership result for pool %d on chain %s: isMember=%t, address=%s", - pool.ID, chainName, resp.IsMember, resp.MemberAddress) - - // Verify response structure - if resp.ChainName != chainName { - t.Errorf("Expected chain name %s, got %s", chainName, resp.ChainName) - } - if resp.PoolID != pool.ID { - t.Errorf("Expected pool ID %d, got %d", pool.ID, resp.PoolID) - } - - // For skale chain and pool 1, we expect membership to be true - if chainName == "skale" && pool.ID == 1 { - if !resp.IsMember { - t.Errorf("Expected membership to be true for pool 1 on skale chain") - } - if resp.MemberAddress == "0x0000000000000000000000000000000000000000" { - t.Errorf("Expected non-zero member address for pool 1 on skale chain") - } - } - }) - } - }) - } -} - // TestPoolZeroHandling tests that pool 0 is properly handled func TestPoolZeroHandling(t *testing.T) { bl, err := NewFxBlockchain( @@ -130,51 +43,3 @@ func TestPoolZeroHandling(t *testing.T) { t.Logf("Pool 0 handling test passed: isMember=%t, address=%s", resp.IsMember, resp.MemberAddress) } - -// TestPoolDiscoveryReturnsCorrectPoolIDs tests that pool discovery returns the correct pool IDs -func TestPoolDiscoveryReturnsCorrectPoolIDs(t *testing.T) { - bl, err := NewFxBlockchain( - NewSimpleKeyStorer(""), - WithTimeout(30), - ) - if err != nil { - t.Fatalf("Failed to create blockchain: %v", err) - } - - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - - chains := []string{"skale", "base"} - - for _, chainName := range chains { - t.Run("PoolIDs_"+chainName, func(t *testing.T) { - poolList, err := bl.HandleEVMPoolList(ctx, chainName) - if err != nil { - t.Fatalf("Pool discovery failed: %v", err) - } - - t.Logf("Chain %s discovered pools:", chainName) - for i, pool := range poolList.Pools { - t.Logf(" Pool[%d]: ID=%d, Name=%s, Creator=%s", i, pool.ID, pool.Name, pool.Creator) - - // Verify pool ID is valid (not 0) - if pool.ID == 0 { - t.Errorf("Pool at index %d has invalid ID 0", i) - } - } - - // We expect to find pool 1 on both chains - foundPool1 := false - for _, pool := range poolList.Pools { - if pool.ID == 1 { - foundPool1 = true - break - } - } - - if !foundPool1 { - t.Errorf("Expected to find pool ID 1 on chain %s", chainName) - } - }) - } -} diff --git a/blockchain/pool_test.go b/blockchain/pool_test.go index 6fab6fcb..2f18c766 100644 --- a/blockchain/pool_test.go +++ b/blockchain/pool_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -315,7 +316,7 @@ func TestFetchUsersAndPopulateSets(t *testing.T) { ctx := context.Background() // Test fetching users for a pool - err = bl.FetchUsersAndPopulateSets(ctx, "1", true, 30*time.Second) + _ = bl.FetchUsersAndPopulateSets(ctx, "1", true, 30*time.Second) // This may fail without proper setup, but tests the interface // The method should handle the case gracefully @@ -343,7 +344,7 @@ func TestDiscoverPoolAndChain(t *testing.T) { // Test discovery functionality // This is a private method, so we test it indirectly through FetchUsersAndPopulateSets - err = bl.FetchUsersAndPopulateSets(ctx, "0", true, 30*time.Second) + _ = bl.FetchUsersAndPopulateSets(ctx, "0", true, 30*time.Second) // The method should handle discovery gracefully assert.NotNil(t, bl) @@ -404,29 +405,37 @@ func TestMembershipCheck(t *testing.T) { func TestPoolDiscoveryWithMockServer(t *testing.T) { ctx := context.Background() - // Create mock server that returns the real contract responses + // Create mock server that returns the real contract responses. It runs on the server's goroutines, so it + // reports problems with t.Errorf (require/t.Fatalf must only be called from the test goroutine). mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var requestBody map[string]interface{} - err := json.NewDecoder(r.Body).Decode(&requestBody) - require.NoError(t, err) - - method, ok := requestBody["method"].(string) - require.True(t, ok) + if err := json.NewDecoder(r.Body).Decode(&requestBody); err != nil { + t.Errorf("mock: decoding request: %v", err) + return + } + method, _ := requestBody["method"].(string) switch method { case "eth_call": - params, ok := requestBody["params"].([]interface{}) - require.True(t, ok) - require.Len(t, params, 2) - - callParams, ok := params[0].(map[string]interface{}) - require.True(t, ok) - - data, ok := callParams["data"].(string) - require.True(t, ok) + params, _ := requestBody["params"].([]interface{}) + if len(params) != 2 { + t.Errorf("mock: eth_call wants 2 params, got %d", len(params)) + return + } + callParams, _ := params[0].(map[string]interface{}) + data, _ := callParams["data"].(string) // Mock responses based on the method being called - if data == abi.EncodePoolsCall(1) { + if data == abi.EncodePoolIdsCall(0) { + // One pool on the chain: poolIds(0) = 1; any later index reverts below. + response := map[string]interface{}{ + "jsonrpc": "2.0", + "id": requestBody["id"], + "result": "0x0000000000000000000000000000000000000000000000000000000000000001", + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(response) + } else if data == abi.EncodePoolsCall(1) { // Return the real pools(1) response response := map[string]interface{}{ "jsonrpc": "2.0", @@ -458,10 +467,13 @@ func TestPoolDiscoveryWithMockServer(t *testing.T) { json.NewEncoder(w).Encode(response) } default: - t.Fatalf("Unexpected method: %s", method) + t.Errorf("mock: unexpected method %q", method) } })) defer mockServer.Close() + skale := GetChainConfigs()["skale"] + skale.RPC = mockServer.URL + skale.BackupRPC = "" // Create test peer priv, _, err := crypto.GenerateKeyPairWithReader(crypto.RSA, 2048, rand.Reader) @@ -478,8 +490,10 @@ func TestPoolDiscoveryWithMockServer(t *testing.T) { WithAuthorizer(h.ID()), WithTimeout(30), WithBlockchainEndPoint(mockServer.URL), + withChainConfigs(map[string]ChainConfig{"skale": skale}), ) require.NoError(t, err) + t.Cleanup(func() { _ = bl.Shutdown(context.Background()) }) // Test HandleEVMPoolList response, err := bl.HandleEVMPoolList(ctx, "skale") @@ -506,7 +520,8 @@ func TestPoolDiscoveryWithMockServer(t *testing.T) { memberResponse, err := bl.HandleIsMemberOfPool(ctx, memberReq) require.NoError(t, err, "HandleIsMemberOfPool should succeed") assert.True(t, memberResponse.IsMember, "Peer should be a member of pool 1") - assert.Equal(t, "0xCe12f8cE914dA115191De28f2E1796a24E475B72", memberResponse.MemberAddress, "Member address should match") + assert.True(t, strings.EqualFold("0xCe12f8cE914dA115191De28f2E1796a24E475B72", memberResponse.MemberAddress), + "Member address should match, got %s", memberResponse.MemberAddress) t.Logf("Membership check test successful:") t.Logf(" PeerID: %s", memberReq.PeerID) diff --git a/blockchain/proxy.go b/blockchain/proxy.go index 485e27da..bf54f0ea 100644 --- a/blockchain/proxy.go +++ b/blockchain/proxy.go @@ -21,7 +21,7 @@ const ( // blockchain commands forwarded by kubo's libp2p stream mounting. // Requests are authenticated via signed headers (see auth_signed.go). func (bl *FxBlockchain) StartProxy(ctx context.Context) error { - listener, err := net.Listen("tcp", ProxyListenAddr) + listener, err := net.Listen("tcp", bl.proxyListenAddr) if err != nil { return err } @@ -40,7 +40,7 @@ func (bl *FxBlockchain) StartProxy(ctx context.Context) error { log.Errorw("Proxy server stopped erroneously", "err", err) } }() - log.Infow("Blockchain proxy server started", "addr", ProxyListenAddr) + log.Infow("Blockchain proxy server started", "addr", bl.proxyListenAddr) return nil } diff --git a/blox/blox.go b/blox/blox.go index 581d059a..6b4f866c 100644 --- a/blox/blox.go +++ b/blox/blox.go @@ -17,7 +17,6 @@ import ( "github.com/functionland/go-fula/blockchain" "github.com/functionland/go-fula/common" "github.com/ipfs-cluster/ipfs-cluster/api" - "github.com/ipfs/boxo/path" "github.com/ipfs/go-cid" logging "github.com/ipfs/go-log/v2" "github.com/ipfs/kubo/client/rpc" @@ -112,27 +111,6 @@ func New(o ...Option) (*Blox, error) { return &p, nil } -func (p *Blox) storeCidIPFS(ctx context.Context, c path.Path) error { - getCtx, cancel := context.WithTimeout(ctx, 5*time.Second) - defer cancel() - if p.rpc == nil { - return fmt.Errorf("IPFS rpc is undefined") - } - _, err := p.rpc.Block().Get(getCtx, c) - if err != nil { - log.Errorw("It seems that the link is not found", "c", c, "err", err) - return err - } - pinCtx, cancel := context.WithTimeout(ctx, 5*time.Second) - defer cancel() - err = p.rpc.Pin().Add(pinCtx, c) - if err != nil { - log.Errorw("It seems that the link is found but not pinned", "c", c, "err", err) - return err - } - return nil -} - func (p *Blox) StoreCid(ctx context.Context, l ipld.Link, limit int) error { // Data replication has moved to kubo/IPFS-cluster. // StoreCid via exchange pull is no longer supported. @@ -664,7 +642,9 @@ func (p *Blox) Start(ctx context.Context) error { log.Errorf("Error retrieving last checked time: %v", err) continue } - p.topicName = p.getPoolName() + if p.getPoolName != nil { + p.topicName = p.getPoolName() + } if p.topicName != "0" { shortCtx, shortCtxCancel := context.WithDeadline(p.ctx, time.Now().Add(60*time.Second)) defer shortCtxCancel() diff --git a/blox/blox_test.go b/blox/blox_test.go index 02341901..8c86a19e 100644 --- a/blox/blox_test.go +++ b/blox/blox_test.go @@ -169,7 +169,7 @@ func TestBloxStoreCid(t *testing.T) { require.NoError(t, err) // Test StoreCid operation (this will likely fail due to network setup) - err = blox.StoreCid(ctx, link, 3) // Limit of 3 replicas + _ = blox.StoreCid(ctx, link, 3) // Limit of 3 replicas // We don't assert success here as it depends on network connectivity // But we test that the method exists and can be called assert.NotNil(t, blox) @@ -216,7 +216,7 @@ func TestBloxStoreManifest(t *testing.T) { } // Test StoreManifest operation - err = blox.StoreManifest(ctx, linksWithLimits, 10) + _ = blox.StoreManifest(ctx, linksWithLimits, 10) // We don't assert success here as it depends on network connectivity // But we test that the method exists and can be called assert.NotNil(t, blox) @@ -260,11 +260,11 @@ func TestBloxPushPull(t *testing.T) { defer h2.Close() // Test Push operation (will likely fail due to network setup) - err = blox.Push(ctx, h2.ID(), link) + _ = blox.Push(ctx, h2.ID(), link) // We don't assert success here as it depends on network connectivity // Test Pull operation (will likely fail due to network setup) - err = blox.Pull(ctx, h2.ID(), link) + _ = blox.Pull(ctx, h2.ID(), link) // We don't assert success here as it depends on network connectivity assert.NotNil(t, blox) diff --git a/blox/kubo_proxy.go b/blox/kubo_proxy.go index 6493d11b..0b2cd085 100644 --- a/blox/kubo_proxy.go +++ b/blox/kubo_proxy.go @@ -29,8 +29,8 @@ const ( healthCheckInterval = 30 * time.Second - poolsAPIEndpoint = "https://pools.fx.land/pools/" - serverKuboPeerIDCachePath = "/internal/.tmp/pool_%s_server_kubo.tmp" + poolsAPIEndpoint = "https://pools.fx.land/pools/" + serverKuboPeerIDCachePath = "/internal/.tmp/pool_%s_server_kubo.tmp" // Static relay PeerID — matches Peering.Peers / Swarm.RelayClient.StaticRelays // in the kubo template config. The watchdog gates fula restarts on actual @@ -127,17 +127,15 @@ func resolveProxyTargetIP() string { if ifaces == nil { ifaces, _ = net.Interfaces() } - if ifaces != nil { - for _, iface := range ifaces { - if iface.Name == "lo" || strings.HasPrefix(iface.Name, "docker") || - strings.HasPrefix(iface.Name, "br-") || strings.HasPrefix(iface.Name, "veth") { - continue - } - if ip := getIPv4FromInterface(iface.Name); ip != "" { - log.Infow("Detected container bridge IP for proxy target (bridge networking mode)", - "iface", iface.Name, "ip", ip) - return ip - } + for _, iface := range ifaces { + if iface.Name == "lo" || strings.HasPrefix(iface.Name, "docker") || + strings.HasPrefix(iface.Name, "br-") || strings.HasPrefix(iface.Name, "veth") { + continue + } + if ip := getIPv4FromInterface(iface.Name); ip != "" { + log.Infow("Detected container bridge IP for proxy target (bridge networking mode)", + "iface", iface.Name, "ip", ip) + return ip } } @@ -433,16 +431,6 @@ func (p *Blox) watchKuboP2P(ctx context.Context) { continue } - // If a previous bounce already verified as failed, count it. - // We judge a bounce as failed if circuit is still missing - // bounceVerifyGracePeriod after the bounce. - if !lastRestartAt.IsZero() && time.Since(lastRestartAt) >= bounceVerifyGracePeriod { - // Only count once per failed bounce (next gate-1 reset clears it). - if consecutiveFailedRestarts == 0 || lastRestartAt.After(time.Now().Add(-bounceVerifyGracePeriod-minTimeBetweenFulaRestarts)) { - // no-op; we account for it on the actual bounce decision below - } - } - // Gate 3: short cooldown between back-to-back bounces. if time.Since(lastRestartAt) < minTimeBetweenFulaRestarts { log.Infow("Skipping fula restart — cooldown not elapsed", diff --git a/cmd/blox/main.go b/cmd/blox/main.go index 13a665b1..5ce4376b 100644 --- a/cmd/blox/main.go +++ b/cmd/blox/main.go @@ -612,54 +612,6 @@ func updateChainName(newChainName string) error { return nil } -func updateConfig(p []peer.ID) error { - configMu.Lock() - defer configMu.Unlock() - // Load existing config file - configData, err := os.ReadFile(app.configPath) - if err != nil { - return err - } - - // Parse the existing config file - if err := yaml.Unmarshal(configData, &app.config); err != nil { - return err - } - - // Create a map to hold unique peer IDs - uniquePeers := make(map[string]bool) - - // Add existing AuthorizedPeers to the map - for _, pidStr := range app.config.AuthorizedPeers { - uniquePeers[pidStr] = true - } - - // Convert the slice of peer.ID to a slice of strings - for _, pid := range p { - // Convert the peer.ID to string - pidStr := pid.String() - // Check if the peer.ID is already in the map - if !uniquePeers[pidStr] { - // If it's not in the map, add it to the map and the slice - uniquePeers[pidStr] = true - app.config.AuthorizedPeers = append(app.config.AuthorizedPeers, pidStr) - } - } - - logger.Infof("Authorized peers: %v", app.config.AuthorizedPeers) - // Write back the updated config to the file - configData, err = yaml.Marshal(app.config) - if err != nil { - return err - } - - if err := os.WriteFile(app.configPath, configData, 0700); err != nil { - return err - } - - return nil -} - func updatePoolName(newPoolName string) error { configMu.Lock() defer configMu.Unlock() @@ -879,12 +831,16 @@ func CustomStorageWriteOpenerNone(lctx linking.LinkContext) (io.Writer, ipld.Blo // mostly useful to hook up things that register in the default muxer, // and don't provide a convenient http.Handler entry point, such as // expvar and http/pprof. +// +//lint:ignore U1000 kept unused on purpose: it is the only importer of kubo's commands/corehttp stack, and dropping that removes ~250 packages and their init()s (e.g. the dag-jose and git codec registrations) from the daemon func defaultMux(path string) corehttp.ServeOption { return func(node *core.IpfsNode, _ net.Listener, mux *http.ServeMux) (*http.ServeMux, error) { mux.Handle(path, http.DefaultServeMux) return mux, nil } -} // serveHTTPApi collects options, creates listener, prints status message and starts serving requests. +} + +//lint:ignore U1000 kept unused on purpose: it is the only importer of kubo's commands/corehttp stack, and dropping that removes ~250 packages and their init()s (e.g. the dag-jose and git codec registrations) from the daemon func rewriteMaddrToUseLocalhostIfItsAny(maddr ma.Multiaddr) ma.Multiaddr { first, rest := ma.SplitFirst(maddr) @@ -897,6 +853,10 @@ func rewriteMaddrToUseLocalhostIfItsAny(maddr ma.Multiaddr) ma.Multiaddr { return maddr // not ip } } + +// serveHTTPApi collects options, creates listener, prints status message and starts serving requests. +// +//lint:ignore U1000 kept unused on purpose: it is the only importer of kubo's commands/corehttp stack, and dropping that removes ~250 packages and their init()s (e.g. the dag-jose and git codec registrations) from the daemon func serveHTTPApi(cctx *oldcmds.Context) (<-chan error, error) { cfg, err := cctx.GetConfig() if err != nil { diff --git a/cmd/servertest/main.go b/cmd/servertest/main.go index 1bd1cff0..8b86b5b3 100644 --- a/cmd/servertest/main.go +++ b/cmd/servertest/main.go @@ -200,7 +200,7 @@ func main() { const poolName = "1" ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) - //ctx = network.WithUseTransient(ctx, "fx.exchange") + //ctx = network.WithAllowLimitedConn(ctx, "fx.exchange") defer cancel() // Elevate log level to show internal communications. diff --git a/exchange/example_test.go b/exchange/example_test.go deleted file mode 100644 index cbff63cb..00000000 --- a/exchange/example_test.go +++ /dev/null @@ -1,837 +0,0 @@ -package exchange_test - -import ( - "bytes" - "context" - "encoding/base64" - "encoding/json" - "fmt" - "io" - "net/http" - "time" - - "github.com/functionland/go-fula/blox" - logging "github.com/ipfs/go-log/v2" - "github.com/ipld/go-ipld-prime/codec/dagjson" - "github.com/ipld/go-ipld-prime/fluent" - "github.com/ipld/go-ipld-prime/node/basicnode" - "github.com/libp2p/go-libp2p" - "github.com/libp2p/go-libp2p/core/crypto" - "github.com/libp2p/go-libp2p/core/peer" - "github.com/libp2p/go-libp2p/core/peerstore" -) - -var log = logging.Logger("fula/dhttest") - -// requestLoggerMiddleware logs the details of each request -func requestLoggerMiddleware(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - // Log the request - body, _ := io.ReadAll(r.Body) - log.Debugw("Received request", "url", r.URL.Path, "method", r.Method, "body", string(body)) - if r.URL.Path == "/fula/pool/vote" { - fmt.Printf("Voted on QmUg1bGBZ1rSNt3LZR7kKf9RDy3JtJLZZDZGKrzSP36TMe %s", string(body)) - } - - // Create a new io.Reader from the read body as the original body is now drained - r.Body = io.NopCloser(bytes.NewBuffer(body)) - - // Call the next handler - next.ServeHTTP(w, r) - }) -} -func startMockServer(addr string) *http.Server { - handler := http.NewServeMux() - - handler.HandleFunc("/fula/pool/join", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "pool_id": 1, - "account": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/pool/cancel_join", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "pool_id": 1, - "account": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/pool/poolrequests", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "poolrequests": []map[string]interface{}{ - { - "pool_id": 1, - "account": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - "voted": []string{}, - "positive_votes": 0, - "peer_id": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - }, - }, - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/pool/all", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "pools": []map[string]interface{}{ - { - "pool_id": 1, - "creator": "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", - "pool_name": "PoolTest1", - "region": "Ontario", - "parent": nil, - "participants": []string{ - "12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM", - "12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX", - "12D3KooWRde3N9rHE8vEyzTiPMVBvs1RpjS4oaWjVkfAt17412vX", - }, - }, - }, - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/pool/users", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "users": []map[string]interface{}{ - { - "account": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - "pool_id": nil, - "request_pool_id": 1, - "peer_id": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - }, - { - "account": "12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM", - "pool_id": 1, - "request_pool_id": nil, - "peer_id": "12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM", - }, - { - "account": "12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX", - "pool_id": 1, - "request_pool_id": nil, - "peer_id": "12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX", - }, - { - "account": "12D3KooWRde3N9rHE8vEyzTiPMVBvs1RpjS4oaWjVkfAt17412vX", - "pool_id": 1, - "request_pool_id": nil, - "peer_id": "12D3KooWRde3N9rHE8vEyzTiPMVBvs1RpjS4oaWjVkfAt17412vX", - }, - }, - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/pool/vote", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "pool_id": 1, - "account": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/manifest/available", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "manifests": []map[string]interface{}{ - { - "pool_id": 1, - "manifest_metadata": map[string]interface{}{ - "job": map[string]string{ - "engine": "IPFS", - "uri": "bafyreidulpo7on77a6pkq7c6da5mlj4n2p3av2zjomrpcpeht5zqgafc34", - "work": "Storage", - }, - }, - "replication_available": 2, - }, - { - "pool_id": 1, - "manifest_metadata": map[string]interface{}{ - "job": map[string]string{ - "engine": "IPFS", - "uri": "bafyreibzsetfhqrayathm5tkmm7axuljxcas3pbqrncrosx2fiky4wj5gy", - "work": "Storage", - }, - }, - "replication_available": 1, - }, - }, - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/pool/leave", func(w http.ResponseWriter, r *http.Request) { - response := map[string]interface{}{ - "pool_id": 1, - "account": "12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q", - } - json.NewEncoder(w).Encode(response) - }) - - handler.HandleFunc("/fula/manifest/batch_storage", func(w http.ResponseWriter, r *http.Request) { - var reqBody struct { - CIDs []string `json:"cid"` - PoolID int `json:"pool_id"` - } - - // Decode the JSON body of the request - if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - // Make sure to close the request body - defer r.Body.Close() - - // Use the CIDs from the request in the response - response := map[string]interface{}{ - "pool_id": reqBody.PoolID, - "cid": reqBody.CIDs, - } - - // Encode the response as JSON - if err := json.NewEncoder(w).Encode(response); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - }) - - // Wrap the handlers with the logging middleware - loggedHandler := requestLoggerMiddleware(handler) - - // Create an HTTP server - server := &http.Server{ - Addr: addr, - Handler: loggedHandler, - } - // Start the server in a new goroutine - go func() { - if err := server.ListenAndServe(); err != http.ErrServerClosed { - panic(err) // Handle the error as you see fit - } - }() - - // Give the server a moment to start - time.Sleep(time.Millisecond * 100) - - return server -} - -func generateIdentity(id int) crypto.PrivKey { - var pid crypto.PrivKey - switch id { - case 1: //12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - key1 := "CAESQJ5GGDgYMGs8eWNCSotGC/qnuw3pfwtG6XcAumHc4CR33IrywkIsmSlMOK7RdP78RgFmYgyrZxz7fP1xux0I88w=" - km1, err := base64.StdEncoding.DecodeString(key1) - if err != nil { - panic(err) - } - pid, err = crypto.UnmarshalPrivateKey(km1) - if err != nil { - panic(err) - } - - case 2: //12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX - key2 := "CAESQHSuiy3FbrTSh7MzXI6coF52bTXrtx3ZorFzIbKnZeBAbQGC6PMp90hKgAiM4yW5/TkRBQhqgPN99AwdiLOS27Q=" - km2, err := base64.StdEncoding.DecodeString(key2) - if err != nil { - panic(err) - } - pid, err = crypto.UnmarshalPrivateKey(km2) - if err != nil { - panic(err) - } - - case 3: //12D3KooWRde3N9rHE8vEyzTiPMVBvs1RpjS4oaWjVkfAt17412vX - key3 := "CAESQHAfwsoKLRHraOpYeV6DBjWeG4B9PpSWLyMym2modqej6vuMoMJ5FiA1ivOyihgJxeqKsVue/9cjKlxSNoMQCrQ=" - km3, err := base64.StdEncoding.DecodeString(key3) - if err != nil { - panic(err) - } - pid, err = crypto.UnmarshalPrivateKey(km3) - if err != nil { - panic(err) - } - - case 4: //12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q - key4 := "CAESQCKbGJG9XDbfUEMjie3vZYVk9RgXHXCLjTMeBidltp396IK4gNRCMmGbjZeG+ZN4FC+yCLDNB1Vzbg66DaeHvCU=" - km4, err := base64.StdEncoding.DecodeString(key4) - if err != nil { - panic(err) - } - pid, err = crypto.UnmarshalPrivateKey(km4) - if err != nil { - panic(err) - } - } - return pid -} - -func updatePoolName(newPoolName string) error { - return nil -} - -func Example_provideAfterPull() { - server := startMockServer("127.0.0.1:4001") - defer func() { - // Shutdown the server after test - if err := server.Shutdown(context.Background()); err != nil { - panic(err) // Handle the error as you see fit - } - }() - - const poolName = "1" - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - - // Elevate log level to show internal communications. - if err := logging.SetLogLevel("*", "info"); err != nil { - panic(err) - } - - // Use a deterministic random generator to generate deterministic - // output for the example. - - // Instantiate the first node in the pool - h1, err := libp2p.New(libp2p.Identity(generateIdentity(1))) - if err != nil { - panic(err) - } - n1, err := blox.New( - blox.WithPoolName(poolName), - blox.WithTopicName(poolName), - blox.WithUpdatePoolName(updatePoolName), - blox.WithBlockchainEndPoint("127.0.0.1:4001"), - blox.WithRelays([]string{"/dns/relay.dev.fx.land/tcp/4001/p2p/12D3KooWDRrBaAfPwsGJivBoUw5fE7ZpDiyfUjqgiURq2DEcL835"}), - ) - if err != nil { - panic(err) - } - if err := n1.Start(ctx); err != nil { - panic(err) - } - defer n1.Shutdown(ctx) - fmt.Printf("Instantiated node in pool %s with ID: %s\n", poolName, h1.ID().String()) - - // Instantiate the second node in the pool - h2, err := libp2p.New(libp2p.Identity(generateIdentity(2))) - if err != nil { - panic(err) - } - n2, err := blox.New( - blox.WithPoolName(poolName), - blox.WithTopicName(poolName), - blox.WithUpdatePoolName(updatePoolName), - blox.WithBlockchainEndPoint("127.0.0.1:4001"), - blox.WithRelays([]string{"/dns/relay.dev.fx.land/tcp/4001/p2p/12D3KooWDRrBaAfPwsGJivBoUw5fE7ZpDiyfUjqgiURq2DEcL835"}), - ) - if err != nil { - panic(err) - } - if err := n2.Start(ctx); err != nil { - panic(err) - } - defer n2.Shutdown(ctx) - fmt.Printf("Instantiated node in pool %s with ID: %s\n", poolName, h2.ID().String()) - - // Instantiate the third node in the pool - h3, err := libp2p.New(libp2p.Identity(generateIdentity(3))) - if err != nil { - panic(err) - } - n3, err := blox.New( - blox.WithPoolName(poolName), - blox.WithTopicName(poolName), - blox.WithUpdatePoolName(updatePoolName), - blox.WithBlockchainEndPoint("127.0.0.1:4001"), - blox.WithRelays([]string{"/dns/relay.dev.fx.land/tcp/4001/p2p/12D3KooWDRrBaAfPwsGJivBoUw5fE7ZpDiyfUjqgiURq2DEcL835"}), - ) - if err != nil { - panic(err) - } - if err := n3.Start(ctx); err != nil { - panic(err) - } - defer n3.Shutdown(ctx) - fmt.Printf("Instantiated node in pool %s with ID: %s\n", poolName, h3.ID().String()) - - // Instantiate the fourth node not in the pool - h4, err := libp2p.New(libp2p.Identity(generateIdentity(4))) - if err != nil { - panic(err) - } - n4, err := blox.New( - blox.WithPoolName("0"), - blox.WithTopicName("0"), - blox.WithUpdatePoolName(updatePoolName), - blox.WithBlockchainEndPoint("127.0.0.1:4001"), - blox.WithRelays([]string{"/dns/relay.dev.fx.land/tcp/4001/p2p/12D3KooWDRrBaAfPwsGJivBoUw5fE7ZpDiyfUjqgiURq2DEcL835"}), - ) - if err != nil { - panic(err) - } - if err := n4.Start(ctx); err != nil { - panic(err) - } - defer n4.Shutdown(ctx) - fmt.Printf("Instantiated node in pool %s with ID: %s\n", "0", h4.ID().String()) - - // Connect n1 to n2 and n3 so that there is a path for gossip propagation. - // Note that we are not connecting n2 to n3 as they should discover - // each other via pool's iexist announcements. - h1.Peerstore().AddAddrs(h2.ID(), h2.Addrs(), peerstore.PermanentAddrTTL) - if err = h1.Connect(ctx, peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()}); err != nil { - panic(err) - } - h1.Peerstore().AddAddrs(h3.ID(), h3.Addrs(), peerstore.PermanentAddrTTL) - if err = h1.Connect(ctx, peer.AddrInfo{ID: h3.ID(), Addrs: h3.Addrs()}); err != nil { - panic(err) - } - - // Wait until the nodes discover each other - for { - if len(h1.Peerstore().Peers()) >= 3 && - len(h2.Peerstore().Peers()) >= 3 && - len(h3.Peerstore().Peers()) >= 3 { - break - } - select { - case <-ctx.Done(): - panic(ctx.Err()) - default: - time.Sleep(time.Second) - } - } - - fmt.Printf("Finally %s peerstore contains >=3 nodes:\n", h1.ID()) - - fmt.Printf("Finally %s peerstore contains >=3 nodes:\n", h2.ID()) - - fmt.Printf("Finally %s peerstore contains >=3 nodes:\n", h3.ID()) - - //Manually adding h4 as it is not in the same pool - h1.Peerstore().AddAddrs(h4.ID(), h4.Addrs(), peerstore.PermanentAddrTTL) - if err = h1.Connect(ctx, peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()}); err != nil { - panic(err) - } - //Manually adding h4 as it is not in the same pool - h2.Peerstore().AddAddrs(h4.ID(), h4.Addrs(), peerstore.PermanentAddrTTL) - if err = h2.Connect(ctx, peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()}); err != nil { - panic(err) - } - //Manually adding h4 as it is not in the same pool - h3.Peerstore().AddAddrs(h4.ID(), h4.Addrs(), peerstore.PermanentAddrTTL) - if err = h3.Connect(ctx, peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()}); err != nil { - panic(err) - } - - // Wait until the fourth node discover others - for { - if len(h4.Peerstore().Peers()) >= 4 { - break - } - select { - case <-ctx.Done(): - panic(ctx.Err()) - default: - time.Sleep(time.Second) - } - } - - //Store a link in h1 and find providers from h2 - - // Generate a sample DAG and store it on node 1 (n1) in the pool, which we will pull from n1 - n1leaf := fluent.MustBuildMap(basicnode.Prototype.Map, 1, func(na fluent.MapAssembler) { - na.AssembleEntry("this").AssignBool(true) - }) - n1leafLink, err := n1.Store(ctx, n1leaf) - if err != nil { - panic(err) - } - n1Root := fluent.MustBuildMap(basicnode.Prototype.Map, 2, func(na fluent.MapAssembler) { - na.AssembleEntry("that").AssignInt(42) - na.AssembleEntry("oneLeafLink").AssignLink(n1leafLink) - }) - n1RootLink, err := n1.Store(ctx, n1Root) - if err != nil { - panic(err) - } - fmt.Printf("%s stored IPLD data with links:\n root: %s\n leaf:%s\n", h1.ID(), n1RootLink, n1leafLink) - - // Generate a sample DAG and store it on node 2 (n1) in the pool, which we will push to n1 - n2leaf := fluent.MustBuildMap(basicnode.Prototype.Map, 1, func(na fluent.MapAssembler) { - na.AssembleEntry("that").AssignBool(false) - }) - n2leafLink, err := n2.Store(ctx, n2leaf) - if err != nil { - panic(err) - } - n2Root := fluent.MustBuildMap(basicnode.Prototype.Map, 2, func(na fluent.MapAssembler) { - na.AssembleEntry("this").AssignInt(24) - na.AssembleEntry("anotherLeafLink").AssignLink(n2leafLink) - }) - n2RootLink, err := n2.Store(ctx, n2Root) - if err != nil { - panic(err) - } - fmt.Printf("%s stored IPLD data with links:\n root: %s\n leaf:%s\n", h1.ID(), n2RootLink, n2leafLink) - - //n1.UpdateDhtPeers(h2.Peerstore().Peers()) - //n2.UpdateDhtPeers(h1.Peerstore().Peers()) - - err = n1.ProvideLinkByDht(n1RootLink) - if err != nil { - fmt.Print("Error happened in ProvideLinkByDht") - panic(err) - } - peerlist1, err := n2.FindLinkProvidersByDht(n1RootLink) - if err != nil { - fmt.Print("Error happened in FindLinkProvidersByDht") - panic(err) - } - // Iterate over the slice and print the peer ID of each AddrInfo - for _, addrInfo := range peerlist1 { - fmt.Printf("Found %s on %s\n", n1RootLink, addrInfo.ID.String()) // ID.String() converts the peer ID to a string - } - - err = n1.PingDht(h3.ID()) - if err != nil { - fmt.Print("Error happened in PingDht") - panic(err) - } - - fmt.Println("exchanging by Pull...") - // Pull the sample DAG stored on node 1 from node 2 by only asking for the root link. - // Because fetch implementation is recursive, it should fetch the leaf link too. - if err := n2.Pull(ctx, h1.ID(), n1RootLink); err != nil { - panic(err) - } - - // Assert that n2 now has both root and leaf links - if exists, err := n2.Has(ctx, n1RootLink); err != nil { - panic(err) - } else if !exists { - panic("expected n2 to have fetched the entire sample DAG") - } else { - fmt.Printf("%s successfully fetched:\n link: %s\n from %s\n", h2.ID(), n1RootLink, h1.ID()) - n, err := n2.Load(ctx, n1RootLink, basicnode.Prototype.Any) - if err != nil { - panic(err) - } - var buf bytes.Buffer - if err := dagjson.Encode(n, &buf); err != nil { - panic(err) - } - fmt.Printf(" content: %s\n", buf.String()) - } - if exists, err := n2.Has(ctx, n1leafLink); err != nil { - panic(err) - } else if !exists { - panic("expected n2 to have fetched the entire sample DAG") - } else { - fmt.Printf("%s successfully fetched:\n link: %s\n from %s\n", h2.ID(), n1leafLink, h1.ID()) - n, err := n2.Load(ctx, n1leafLink, basicnode.Prototype.Any) - if err != nil { - panic(err) - } - var buf bytes.Buffer - if err := dagjson.Encode(n, &buf); err != nil { - panic(err) - } - fmt.Printf(" content: %s\n", buf.String()) - } - - fmt.Println("exchanging by Push...") - // Push the sample DAG stored on node 2 to node 1 by only pushing the root link. - // Because Push implementation is recursive, it should push the leaf link too. - if err := n2.Push(ctx, h1.ID(), n2leafLink); err != nil { - panic(err) - } - err = n1.ProvideLinkByDht(n2leafLink) - if err != nil { - fmt.Print("Error happened in ProvideLinkByDht n1") - panic(err) - } - err = n2.ProvideLinkByDht(n2leafLink) - if err != nil { - fmt.Print("Error happened in ProvideLinkByDht n2") - panic(err) - } - - peerlist3, err := n3.FindLinkProvidersByDht(n2leafLink) - if err != nil { - fmt.Print("Error happened in FindLinkProvidersByDht3") - panic(err) - } - - // Iterate over the slice and print the peer ID of each AddrInfo - for _, addrInfo := range peerlist3 { - fmt.Printf("Found %s on %s\n", n2leafLink, addrInfo.ID.String()) // ID.String() converts the peer ID to a string - } - - // Unordered output: - // Instantiated node in pool 1 with ID: 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // Instantiated node in pool 1 with ID: 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX - // Instantiated node in pool 1 with ID: 12D3KooWRde3N9rHE8vEyzTiPMVBvs1RpjS4oaWjVkfAt17412vX - // Instantiated node in pool 0 with ID: 12D3KooWRTzN7HfmjoUBHokyRZuKdyohVVSGqKBMF24ZC3tGK78Q - // Finally 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM peerstore contains >=3 nodes: - // Finally 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX peerstore contains >=3 nodes: - // Finally 12D3KooWRde3N9rHE8vEyzTiPMVBvs1RpjS4oaWjVkfAt17412vX peerstore contains >=3 nodes: - // 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM stored IPLD data with links: - // root: bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji - // leaf:bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4 - // 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM stored IPLD data with links: - // root: bafyr4iapi76lbx5zfcjkolrvqbj4fl4jk2uvuz5obwa7xhbjwq3wupgwhm - // leaf:bafyr4iaab3lel4ykjcyzqajx5np2uluetwvfyv3ujupxt5qs57owhpo6ty - // Found bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji on 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // exchanging by Pull... - // 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX successfully fetched: - // link: bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji - // from 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // content: {"oneLeafLink":{"/":"bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4"},"that":42} - // 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX successfully fetched: - // link: bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4 - // from 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // content: {"this":true} - // exchanging by Push... - // Found bafyr4iaab3lel4ykjcyzqajx5np2uluetwvfyv3ujupxt5qs57owhpo6ty on 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // Found bafyr4iaab3lel4ykjcyzqajx5np2uluetwvfyv3ujupxt5qs57owhpo6ty on 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX -} - -// Example_poolExchangeDagBetweenPoolNodes starts up a pool with 2 nodes, stores a sample DAG in -// one node and fetches it via GraphSync from the other node. -func Example_poolExchangeDagBetweenPoolNodes() { - server := startMockServer("127.0.0.1:4001") - defer func() { - // Shutdown the server after test - if err := server.Shutdown(context.Background()); err != nil { - panic(err) // Handle the error as you see fit - } - }() - - const poolName = "1" - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - - // Elevate log level to show internal communications. - if err := logging.SetLogLevel("*", "info"); err != nil { - panic(err) - } - - // Use a deterministic random generator to generate deterministic - // output for the example. - - // Instantiate the first node in the pool - h1, err := libp2p.New(libp2p.Identity(generateIdentity(1))) - if err != nil { - panic(err) - } - n1, err := blox.New(blox.WithPoolName(poolName)) - if err != nil { - panic(err) - } - if err := n1.Start(ctx); err != nil { - panic(err) - } - defer n1.Shutdown(ctx) - fmt.Printf("Instantiated node in pool %s with ID: %s\n", poolName, h1.ID().String()) - - // Instantiate the second node in the pool - h2, err := libp2p.New(libp2p.Identity(generateIdentity(2))) - if err != nil { - panic(err) - } - n2, err := blox.New(blox.WithPoolName(poolName)) - if err != nil { - panic(err) - } - if err := n2.Start(ctx); err != nil { - panic(err) - } - defer n2.Shutdown(ctx) - fmt.Printf("Instantiated node in pool %s with ID: %s\n", poolName, h2.ID().String()) - - // Connect n1 to n2. - h1.Peerstore().AddAddrs(h2.ID(), h2.Addrs(), peerstore.PermanentAddrTTL) - if err = h1.Connect(ctx, peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()}); err != nil { - panic(err) - } - h2.Peerstore().AddAddrs(h1.ID(), h1.Addrs(), peerstore.PermanentAddrTTL) - if err = h2.Connect(ctx, peer.AddrInfo{ID: h1.ID(), Addrs: h1.Addrs()}); err != nil { - panic(err) - } - - // Authorize exchange between the two nodes - if err := n1.SetAuth(ctx, h1.ID(), h2.ID(), true); err != nil { - panic(err) - } - if err := n2.SetAuth(ctx, h2.ID(), h1.ID(), true); err != nil { - panic(err) - } - - // Generate a sample DAG and store it on node 1 (n1) in the pool, which we will pull from n1 - n1leaf := fluent.MustBuildMap(basicnode.Prototype.Map, 1, func(na fluent.MapAssembler) { - na.AssembleEntry("this").AssignBool(true) - }) - n1leafLink, err := n1.Store(ctx, n1leaf) - if err != nil { - panic(err) - } - n1Root := fluent.MustBuildMap(basicnode.Prototype.Map, 2, func(na fluent.MapAssembler) { - na.AssembleEntry("that").AssignInt(42) - na.AssembleEntry("oneLeafLink").AssignLink(n1leafLink) - }) - n1RootLink, err := n1.Store(ctx, n1Root) - if err != nil { - panic(err) - } - fmt.Printf("%s stored IPLD data with links:\n root: %s\n leaf:%s\n", h1.ID(), n1RootLink, n1leafLink) - - // Generate a sample DAG and store it on node 2 (n1) in the pool, which we will push to n1 - n2leaf := fluent.MustBuildMap(basicnode.Prototype.Map, 1, func(na fluent.MapAssembler) { - na.AssembleEntry("that").AssignBool(false) - }) - n2leafLink, err := n2.Store(ctx, n2leaf) - if err != nil { - panic(err) - } - n2Root := fluent.MustBuildMap(basicnode.Prototype.Map, 2, func(na fluent.MapAssembler) { - na.AssembleEntry("this").AssignInt(24) - na.AssembleEntry("anotherLeafLink").AssignLink(n2leafLink) - }) - n2RootLink, err := n2.Store(ctx, n2Root) - if err != nil { - panic(err) - } - fmt.Printf("%s stored IPLD data with links:\n root: %s\n leaf:%s\n", h1.ID(), n1RootLink, n1leafLink) - - fmt.Println("exchanging by Pull...") - // Pull the sample DAG stored on node 1 from node 2 by only asking for the root link. - // Because fetch implementation is recursive, it should fetch the leaf link too. - if err := n2.Pull(ctx, h1.ID(), n1RootLink); err != nil { - panic(err) - } - - // Assert that n2 now has both root and leaf links - if exists, err := n2.Has(ctx, n1RootLink); err != nil { - panic(err) - } else if !exists { - panic("expected n2 to have fetched the entire sample DAG, link: " + n1RootLink.String()) - } else { - fmt.Printf("%s successfully fetched:\n link: %s\n from %s\n", h2.ID(), n1RootLink, h1.ID()) - n, err := n2.Load(ctx, n1RootLink, basicnode.Prototype.Any) - if err != nil { - panic(err) - } - var buf bytes.Buffer - if err := dagjson.Encode(n, &buf); err != nil { - panic(err) - } - fmt.Printf(" content: %s\n", buf.String()) - } - if exists, err := n2.Has(ctx, n1leafLink); err != nil { - panic(err) - } else if !exists { - panic("expected n2 to have fetched the entire sample DAG, link: " + n1leafLink.String()) - } else { - fmt.Printf("%s successfully fetched:\n link: %s\n from %s\n", h2.ID(), n1leafLink, h1.ID()) - n, err := n2.Load(ctx, n1leafLink, basicnode.Prototype.Any) - if err != nil { - panic(err) - } - var buf bytes.Buffer - if err := dagjson.Encode(n, &buf); err != nil { - panic(err) - } - fmt.Printf(" content: %s\n", buf.String()) - } - - fmt.Println("exchanging by Push...") - // Push the sample DAG stored on node 2 to node 1 by only pushing the root link. - // Because Push implementation is recursive, it should push the leaf link too. - if err := n2.Push(ctx, h1.ID(), n2RootLink); err != nil { - panic(err) - } - - // Since push is an asynchronous operation, wait until background push is finished - // by periodically checking if link is present on node 1. - for { - if exists, _ := n1.Has(ctx, n2RootLink); exists { - break - } - select { - case <-ctx.Done(): - panic(ctx.Err()) - default: - time.Sleep(time.Second) - } - } - - // Assert that n1 now has both root and leaf links - if exists, err := n1.Has(ctx, n2RootLink); err != nil { - panic(err) - } else if !exists { - panic("expected n2 to have pushed the entire sample DAG") - } else { - fmt.Printf("%s successfully pushed:\n link: %s\n from %s\n", h2.ID(), n1RootLink, h1.ID()) - n, err := n1.Load(ctx, n2RootLink, basicnode.Prototype.Any) - if err != nil { - panic(err) - } - var buf bytes.Buffer - if err := dagjson.Encode(n, &buf); err != nil { - panic(err) - } - fmt.Printf(" content: %s\n", buf.String()) - } - if exists, err := n1.Has(ctx, n2leafLink); err != nil { - panic(err) - } else if !exists { - panic("expected n2 to have pushed the entire sample DAG") - } else { - fmt.Printf("%s successfully pushed:\n link: %s\n from %s\n", h2.ID(), n1leafLink, h1.ID()) - n, err := n1.Load(ctx, n2leafLink, basicnode.Prototype.Any) - if err != nil { - panic(err) - } - var buf bytes.Buffer - if err := dagjson.Encode(n, &buf); err != nil { - panic(err) - } - fmt.Printf(" content: %s", buf.String()) - } - - // Output: - // Instantiated node in pool 1 with ID: 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // Instantiated node in pool 1 with ID: 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX - // 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM stored IPLD data with links: - // root: bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji - // leaf:bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4 - // 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM stored IPLD data with links: - // root: bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji - // leaf:bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4 - // exchanging by Pull... - // 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX successfully fetched: - // link: bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji - // from 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // content: {"oneLeafLink":{"/":"bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4"},"that":42} - // 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX successfully fetched: - // link: bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4 - // from 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // content: {"this":true} - // exchanging by Push... - // 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX successfully pushed: - // link: bafyr4ifwexg2ka3kueem7wp36diai4wzqswkdiqscw2su4llkhgwcmq2ji - // from 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // content: {"anotherLeafLink":{"/":"bafyr4iaab3lel4ykjcyzqajx5np2uluetwvfyv3ujupxt5qs57owhpo6ty"},"this":24} - // 12D3KooWH9swjeCyuR6utzKU1UspiW5RDGzAFvNDwqkT5bUHwuxX successfully pushed: - // link: bafyr4iauqnsshryxfg2262z6mqev5fyef7gmgjk54skmtggnplehusyno4 - // from 12D3KooWQfGkPUkoLDEeJE3H3ZTmu9BZvAdbJpmhha8WpjeSLKMM - // content: {"that":false} -} diff --git a/exchange/exchange_test.go b/exchange/exchange_test.go index 06c02ab8..e6f58f22 100644 --- a/exchange/exchange_test.go +++ b/exchange/exchange_test.go @@ -196,15 +196,15 @@ func TestExchangePushPull(t *testing.T) { link := cidlink.Link{Cid: testCid} // Test Push operation (will likely fail due to network setup) - err = ex.Push(ctx, h2.ID(), link) + _ = ex.Push(ctx, h2.ID(), link) // We don't assert success here as it depends on network connectivity // Test Pull operation (will likely fail due to network setup) - err = ex.Pull(ctx, h2.ID(), link) + _ = ex.Pull(ctx, h2.ID(), link) // We don't assert success here as it depends on network connectivity // Test PullBlock operation - err = ex.PullBlock(ctx, h2.ID(), link) + _ = ex.PullBlock(ctx, h2.ID(), link) // We don't assert success here as it depends on network connectivity assert.NotNil(t, ex) @@ -318,10 +318,10 @@ func TestExchangeErrorHandling(t *testing.T) { link := cidlink.Link{Cid: testCid} // These operations should handle errors gracefully - err = ex.Push(ctx, invalidPeerID, link) + _ = ex.Push(ctx, invalidPeerID, link) // We expect errors but the system should not panic - err = ex.Pull(ctx, invalidPeerID, link) + _ = ex.Pull(ctx, invalidPeerID, link) // We expect errors but the system should not panic assert.NotNil(t, ex) @@ -367,7 +367,7 @@ func TestExchangeTimeout(t *testing.T) { defer h2.Close() // Test operations with timeout context - err = ex.Push(ctx, h2.ID(), link) + _ = ex.Push(ctx, h2.ID(), link) // Should handle timeout gracefully assert.NotNil(t, ex) diff --git a/exchange/fx_exchange.go b/exchange/fx_exchange.go index c1837855..79664833 100644 --- a/exchange/fx_exchange.go +++ b/exchange/fx_exchange.go @@ -323,7 +323,7 @@ func (e *FxExchange) Start(ctx context.Context) error { func (e *FxExchange) Pull(ctx context.Context, from peer.ID, l ipld.Link) error { if e.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.exchange") + ctx = network.WithAllowLimitedConn(ctx, "fx.exchange") } cid := l.(cidlink.Link).Cid @@ -365,7 +365,7 @@ func (e *FxExchange) Pull(ctx context.Context, from peer.ID, l ipld.Link) error } func (e *FxExchange) PullBlock(ctx context.Context, from peer.ID, l ipld.Link) error { if e.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.exchange") + ctx = network.WithAllowLimitedConn(ctx, "fx.exchange") } cl := l.(cidlink.Link).Cid @@ -405,7 +405,7 @@ func (e *FxExchange) PullBlock(ctx context.Context, from peer.ID, l ipld.Link) e func (e *FxExchange) Push(ctx context.Context, to peer.ID, l ipld.Link) error { if e.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.exchange") + ctx = network.WithAllowLimitedConn(ctx, "fx.exchange") } log := log.With("cid", l.(cidlink.Link).Cid) @@ -678,7 +678,7 @@ func (e *FxExchange) handlePull(from peer.ID, w http.ResponseWriter, r *http.Req log.Debug("Instantiating background push in response to pull request") ctx := context.TODO() if e.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.exchange") + ctx = network.WithAllowLimitedConn(ctx, "fx.exchange") } if err := e.Push(ctx, from, cidlink.Link{Cid: p.Link}); err != nil { log.Errorw("Failed to fetch in response to push", "err", err) @@ -757,7 +757,7 @@ func (e *FxExchange) SetAuth(ctx context.Context, on peer.ID, subject peer.ID, a return nil } if e.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.exchange") + ctx = network.WithAllowLimitedConn(ctx, "fx.exchange") } r := authorizationRequest{Subject: subject, Allow: allow} var buf bytes.Buffer diff --git a/mobile/client.go b/mobile/client.go index 94532f7d..867b1037 100644 --- a/mobile/client.go +++ b/mobile/client.go @@ -48,8 +48,8 @@ type Client struct { bloxPid peer.ID relays []string - ipfsDHT *dht.IpfsDHT // Standard IPFS DHT for peer discovery fallback - ipfsDHTReady chan struct{} // Closed when DHT bootstrap completes + ipfsDHT *dht.IpfsDHT // Standard IPFS DHT for peer discovery fallback + ipfsDHTReady chan struct{} // Closed when DHT bootstrap completes ipfsDHTCtx context.Context ipfsDHTCancel context.CancelFunc @@ -112,7 +112,7 @@ func (c *Client) beginOp() (context.Context, func(), error) { // ensureConnected attempts to connect to blox using peerstore addresses (direct + relay), // and falls back to IPFS DHT peer discovery if the direct attempt fails and DHT is enabled. func (c *Client) ensureConnected(ctx context.Context) error { - ctx = network.WithUseTransient(ctx, "fx.mobile") + ctx = network.WithAllowLimitedConn(ctx, "fx.mobile") // Close stale connections to avoid "dial backoff" from expired relay v2 // circuits that libp2p still considers "connected". @@ -196,10 +196,10 @@ func (c *Client) ConnectToBlox() error { // Success is true if at least one ping succeeded. func (c *Client) Ping() ([]byte, error) { type PingResult struct { - Success bool `json:"success"` - Successes int `json:"successes"` - AvgRttMs int64 `json:"avg_rtt_ms"` - Errors []string `json:"errors"` + Success bool `json:"success"` + Successes int `json:"successes"` + AvgRttMs int64 `json:"avg_rtt_ms"` + Errors []string `json:"errors"` } opCtx, done, err := c.beginOp() @@ -214,8 +214,8 @@ func (c *Client) Ping() ([]byte, error) { // Ensure we're connected (direct → relay → DHT) if err := c.ensureConnected(ctx); err != nil { result := PingResult{ - Success: false, - Errors: []string{fmt.Sprintf("connection failed: %v", err)}, + Success: false, + Errors: []string{fmt.Sprintf("connection failed: %v", err)}, } return json.Marshal(result) } @@ -227,7 +227,7 @@ func (c *Client) Ping() ([]byte, error) { for i := 0; i < pingCount; i++ { pingCtx, pingCancel := context.WithTimeout(ctx, 10*time.Second) - pingCtx = network.WithUseTransient(pingCtx, "fx.mobile.ping") + pingCtx = network.WithAllowLimitedConn(pingCtx, "fx.mobile.ping") result := <-libp2pping.Ping(pingCtx, c.h, c.bloxPid) pingCancel() if result.Error != nil { diff --git a/mobile/config.go b/mobile/config.go index 6b13763c..c2d27b70 100644 --- a/mobile/config.go +++ b/mobile/config.go @@ -274,7 +274,7 @@ func (cfg *Config) init(mc *Client) error { } // Allow transient (relay) connections for gostream dials - dialCtx := network.WithUseTransient(ctx, "fx.mobile") + dialCtx := network.WithAllowLimitedConn(ctx, "fx.mobile") // Close stale connections to avoid "dial backoff" from expired relay v2 // circuits that libp2p still considers "connected". diff --git a/mobile/device_test.go b/mobile/device_test.go index d5be1b98..678e3859 100644 --- a/mobile/device_test.go +++ b/mobile/device_test.go @@ -146,7 +146,7 @@ func TestRealDeviceConnection(t *testing.T) { } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() - ctx = network.WithUseTransient(ctx, "test") + ctx = network.WithAllowLimitedConn(ctx, "test") s, err := client.h.NewStream(ctx, client.bloxPid, protocol.ID("/x/fula-blockchain")) if err != nil { t.Fatalf("NewStream failed: %v", err) @@ -173,7 +173,7 @@ func TestRealDeviceConnection(t *testing.T) { } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() - ctx = network.WithUseTransient(ctx, "test") + ctx = network.WithAllowLimitedConn(ctx, "test") conn, err := gostream.Dial(ctx, client.h, client.bloxPid, "/x/fula-blockchain") if err != nil { t.Fatalf("gostream.Dial failed: %v", err) @@ -209,7 +209,7 @@ func TestRealDeviceConnection(t *testing.T) { t.Run("Ping", func(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() - ctx = network.WithUseTransient(ctx, "test") + ctx = network.WithAllowLimitedConn(ctx, "test") const pingCount = 3 var successes int diff --git a/modules/initipfs/main.go b/modules/initipfs/main.go index 16a5cc18..dfb4f8dd 100644 --- a/modules/initipfs/main.go +++ b/modules/initipfs/main.go @@ -9,7 +9,6 @@ import ( "flag" "fmt" "io" - "net/http" "os" "github.com/libp2p/go-libp2p/core/crypto" @@ -128,15 +127,15 @@ type IPFSConfig struct { LowWater int `json:"LowWater"` GracePeriod string `json:"GracePeriod"` } `json:"ConnMgr"` - DisableBandwidthMetrics bool `json:"DisableBandwidthMetrics"` - DisableNatPortMap bool `json:"DisableNatPortMap"` - RelayClient struct { + DisableBandwidthMetrics bool `json:"DisableBandwidthMetrics"` + DisableNatPortMap bool `json:"DisableNatPortMap"` + RelayClient struct { Enabled bool `json:"Enabled,omitempty"` StaticRelays []string `json:"StaticRelays,omitempty"` } `json:"RelayClient"` RelayService struct{} `json:"RelayService"` - ResourceMgr struct{} `json:"ResourceMgr"` - Transports struct { + ResourceMgr struct{} `json:"ResourceMgr"` + Transports struct { Multiplexers map[string]interface{} `json:"Multiplexers"` Network map[string]interface{} `json:"Network"` Security map[string]interface{} `json:"Security"` @@ -226,14 +225,6 @@ func main() { updateIPFSConfigIdentity(&ipfsCfg, config) - /* - // Fetch users that are in the same pool - users := []string{} - if config.PoolName != "0" && config.PoolName != "" { - users = fetchPoolUsers(config.PoolName) - } - updateIPFSConfigBootstrap(&ipfsCfg, config.IpfsBootstrapNodes, users) // We cannot do this as we dont know the ip of these nodes - */ updateDatastorePath(&ipfsCfg, ipfsDatastorePath, apiIpAddr) // Enable libp2p stream mounting so kubo can forward P2P protocols to go-fula's TCP server ipfsCfg.Experimental.Libp2pStreamMounting = true @@ -348,34 +339,6 @@ func readIPFSConfig(sourceIpfsConfig, path string) (IPFSConfig, error) { return cfg, nil } -func fetchPoolUsers(poolName string) []string { - url := "https://api.node3.functionyard.fula.network/fula/pool/users" - payload := map[string]interface{}{ - "pool_id": poolName, - } - payloadBytes, _ := json.Marshal(payload) - resp, err := http.Post(url, "application/json", bytes.NewBuffer(payloadBytes)) - if err != nil { - panic(fmt.Sprintf("Failed to fetch pool users: %v", err)) - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - panic(fmt.Sprintf("Unexpected status code: %d", resp.StatusCode)) - } - - body, _ := io.ReadAll(resp.Body) - var apiResponse ApiResponse - if err := json.Unmarshal(body, &apiResponse); err != nil { - panic(fmt.Sprintf("Failed to parse ApiResponse: %v", err)) - } - - userIDs := []string{} - for _, user := range apiResponse.Users { - userIDs = append(userIDs, user.PeerID) - } - return userIDs -} - // deriveKuboKey derives a separate deterministic Ed25519 key for kubo // from the original private key using HMAC-SHA256 with a fixed domain separator. // This keeps the original key for ipfs-cluster identity while giving kubo @@ -445,11 +408,6 @@ func updateDatastorePath(ipfsCfg *IPFSConfig, newPath string, apiIp string) { ipfsCfg.Addresses.API = "/ip4/" + apiIp + "/tcp/5001" } - -func updateIPFSConfigBootstrap(ipfsCfg *IPFSConfig, predefinedBootstraps, bootstrapPeers []string) { - ipfsCfg.Bootstrap = append(predefinedBootstraps, bootstrapPeers...) -} - func writeIPFSConfig(path string, cfg IPFSConfig) { if cfg.Bootstrap == nil { cfg.Bootstrap = []string{} diff --git a/ping/ping.go b/ping/ping.go index 536561a2..9c3524d8 100644 --- a/ping/ping.go +++ b/ping/ping.go @@ -246,7 +246,7 @@ func (pn *FxPing) Ping(ctx context.Context, to peer.ID) (int, int, error) { } if pn.allowTransientConnection { - ctx = network.WithUseTransient(ctx, "fx.ping") + ctx = network.WithAllowLimitedConn(ctx, "fx.ping") } var totalDuration time.Duration diff --git a/wap/cmd/mdns/mdns.go b/wap/cmd/mdns/mdns.go index d5308948..285b06cc 100644 --- a/wap/cmd/mdns/mdns.go +++ b/wap/cmd/mdns/mdns.go @@ -238,11 +238,11 @@ func NewZeroConfService(port int) (*MDNSServer, error) { log.Debugw("mdns instance name", "name", name) service, err := zeroconf.Register( - name, // unique instance name per device - "_fulatower._tcp", // service type and protocol - "local.", // service domain - port, // service port - meta, // service metadata + name, // unique instance name per device + "_fulatower._tcp", // service type and protocol + "local.", // service domain + port, // service port + meta, // service metadata getLANInterfaces(), // only physical/LAN interfaces ) diff --git a/wap/cmd/mdns/mdns_test.go b/wap/cmd/mdns/mdns_test.go index c00fdf42..b41093b6 100644 --- a/wap/cmd/mdns/mdns_test.go +++ b/wap/cmd/mdns/mdns_test.go @@ -2,7 +2,6 @@ package mdns import ( "context" - "fmt" "sync" "testing" "time" @@ -238,8 +237,3 @@ func keys(m map[string]*zeroconf.ServiceEntry) []string { } return result } - -func init() { - // Suppress log output during tests - _ = fmt.Sprintf("test init") -} diff --git a/wap/pkg/server/lanbind.go b/wap/pkg/server/lanbind.go index 392578f2..949685aa 100644 --- a/wap/pkg/server/lanbind.go +++ b/wap/pkg/server/lanbind.go @@ -165,7 +165,11 @@ func startHotspotWatch(mc *multiCloser, mux http.Handler, listenAddr string) { // Binding is the readiness test: net.Listen fails with "cannot assign requested address" until the interface // holds the IP, so no separate interface polling is needed. func watchForHotspot(stop <-chan struct{}, mc *multiCloser, mux http.Handler, listenAddr string) { + if !mc.trackWatcher() { + return // already closed + } go func() { + defer mc.watchers.Done() ticker := time.NewTicker(hotspotWatchInterval) defer ticker.Stop() for { @@ -300,7 +304,11 @@ func startLANSetupWatch(mc *multiCloser, mux http.Handler, port string, apIP str } handler := lanSetupGuard(withCORS(mux)) + if !mc.trackWatcher() { + return // closed in the meantime + } go func() { + defer mc.watchers.Done() ticker := time.NewTicker(lanWatchInterval) defer ticker.Stop() for { diff --git a/wap/pkg/server/lanbind_test.go b/wap/pkg/server/lanbind_test.go index 22085ccf..0545e3c6 100644 --- a/wap/pkg/server/lanbind_test.go +++ b/wap/pkg/server/lanbind_test.go @@ -33,6 +33,20 @@ func writeConfig(t *testing.T, authorizer string, write bool) { config.FULA_CONFIG_PATH = path } +// claimConfig gives the config file writeConfig created an owner, replacing it atomically (write + rename) at +// the same path, so a running watcher sees the change without config.FULA_CONFIG_PATH changing under it. +func claimConfig(t *testing.T, authorizer string) { + t.Helper() + path := config.FULA_CONFIG_PATH + body := "storeDir: /uniondrive\nauthorizer: " + authorizer + "\n" + if err := os.WriteFile(path+".new", []byte(body), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Rename(path+".new", path); err != nil { + t.Fatal(err) + } +} + func TestBloxHasOwner(t *testing.T) { t.Run("no config file at all → unowned (a fresh box)", func(t *testing.T) { writeConfig(t, "", false) @@ -349,7 +363,7 @@ func TestLANSetupWatchStopsOnceTheBloxHasAnOwner(t *testing.T) { defer func() { _ = mc.Close() }() startLANSetupWatch(mc, http.NewServeMux(), "0", "10.42.0.1", nil) - writeConfig(t, "12D3KooWPnaMDrD7QLZKiT2iktjm9Kucx7XEPrSCUS6TTBbYuiRj", true) // claimed + claimConfig(t, "12D3KooWPnaMDrD7QLZKiT2iktjm9Kucx7XEPrSCUS6TTBbYuiRj") // claimed mu.Lock() addrs = append(addrs, addr) mu.Unlock() diff --git a/wap/pkg/server/server.go b/wap/pkg/server/server.go index 349c77c1..e988b562 100644 --- a/wap/pkg/server/server.go +++ b/wap/pkg/server/server.go @@ -55,6 +55,21 @@ type multiCloser struct { // context: that one is `defer cancel()`-ed, so it dies the moment Serve returns, and a watcher hung off // it would exit on its first tick without ever retrying. The watcher lives until the server is closed. stopWatch chan struct{} + // watchers counts the running background watchers (hotspot, LAN setup) so Close() returns only once they + // have stopped; nothing they read can then change under them. + watchers sync.WaitGroup +} + +// trackWatcher registers a background watcher with Close(), or reports false if the server is already closed +// (the watcher must then not start). Called under mc.mu, so it can't race Close()'s Wait. +func (mc *multiCloser) trackWatcher() bool { + mc.mu.Lock() + defer mc.mu.Unlock() + if mc.closed { + return false + } + mc.watchers.Add(1) + return true } // add registers a listener, or closes it immediately and reports false if the server is already shutting down @@ -70,10 +85,9 @@ func (mc *multiCloser) add(l io.Closer) bool { return true } -// Implement Close method for multiCloser +// Close closes every listener, stops the background watchers and waits for them to exit. func (mc *multiCloser) Close() error { mc.mu.Lock() - defer mc.mu.Unlock() mc.closed = true if mc.stopWatch != nil { close(mc.stopWatch) @@ -86,6 +100,10 @@ func (mc *multiCloser) Close() error { } } mc.listeners = nil + mc.mu.Unlock() + // Wait without holding mu: a watcher may be about to call add(), which takes it. With closed set, add() + // refuses and the watcher returns, so this is at most one tick's work. + mc.watchers.Wait() return err } @@ -621,68 +639,6 @@ func disableAccessPointHandler(w http.ResponseWriter, r *http.Request) { } } -func getIPFromSpecificNetwork(ctx context.Context, ssid string) (string, error) { - ifaces, err := net.Interfaces() - if err != nil { - return "", err - } - - for _, iface := range ifaces { - var cmdString string - cmdString = "iw dev " + iface.Name + " link" - - out, _, err := wifi.RunCommand(ctx, cmdString) - - // Check network connection based on the command used - var isConnectedToNetwork bool - if err == nil { - isConnectedToNetwork = strings.Contains(out, ssid) - } - - // If the interface is connected to the specified network, return its IP - if isConnectedToNetwork { - addrs, err := iface.Addrs() - if err != nil { - return "", err - } - - for _, addr := range addrs { - ip, _, err := net.ParseCIDR(addr.String()) - if err != nil { - continue - } - - if !ip.IsLoopback() && ip.To4() != nil { - return ip.String(), nil - } - } - } - } - - return "", fmt.Errorf("no non-loopback IP address found for network %s", ssid) -} - -// This finds the ip address of the device -func getNonLoopbackIP() (string, error) { - addrs, err := net.InterfaceAddrs() - if err != nil { - return "", err - } - - for _, addr := range addrs { - ip, _, err := net.ParseCIDR(addr.String()) - if err != nil { - continue - } - - if !ip.IsLoopback() && ip.To4() != nil { - return ip.String(), nil - } - } - - return "", fmt.Errorf("no non-loopback IP address found") -} - func joinPoolHandler(w http.ResponseWriter, r *http.Request) { var poolID string diff --git a/wap/pkg/wifi/properties.go b/wap/pkg/wifi/properties.go index fd1f7d33..b0727c18 100644 --- a/wap/pkg/wifi/properties.go +++ b/wap/pkg/wifi/properties.go @@ -177,7 +177,7 @@ func GetDockerImageBuildDates() (GetDockerImageBuildDatesResponse, error) { name = strings.TrimPrefix(c.Names[0], "/") } - imageInspect, _, err := cli.ImageInspectWithRaw(context.Background(), c.ImageID) + imageInspect, err := cli.ImageInspect(context.Background(), c.ImageID) if err != nil { continue } @@ -544,7 +544,7 @@ func GetContainerInfo(containerName string) (DockerInfo, error) { return DockerInfo{}, err } - imageJSON, _, err := cli.ImageInspectWithRaw(context.Background(), containerJSON.Image) + imageJSON, err := cli.ImageInspect(context.Background(), containerJSON.Image) if err != nil { return DockerInfo{}, err }