Ghost peers handling

.github/workflows/release.yml

@@ -6,6 +6,37 @@ on:
       - 'v*'   # Run when pushing a version tag like v1.0.0
 
 jobs:
+  burn-version:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout the repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Get release version
+        id: get_version
+        run: |
+          echo "RELEASE_TAG=${GITHUB_REF#refs/tags/}" >> $GITHUB_ENV
+
+      - name: Checkout release tag
+        run: |
+          git checkout tags/${{ env.RELEASE_TAG }} -b release-version-update
+
+      - name: Update version in code
+        run: |
+          # Replace in a specific file — adjust pattern and file path
+          sed -i "s/0\.0\.0-dev\.0/${{ env.RELEASE_TAG }}/g" src/internal/protocol/host.go
+
+      - name: Commit and push changes
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          git add src/internal/protocol/host.go
+          git commit -m "Update version to ${{ env.RELEASE_TAG }}"
+          git tag -d ${{ env.RELEASE_TAG }}
+          git tag -f ${{ env.RELEASE_TAG }}
+          git push --force origin refs/tags/${{ env.RELEASE_TAG }}
   build:
     name: Build Go Binary
     runs-on: ubuntu-latest
@@ -14,6 +45,16 @@ jobs:
       - name: Check out source code
         uses: actions/checkout@v4
 
+      - name: Get release version
+        id: get_version
+        run: |
+          echo "RELEASE_TAG=${GITHUB_REF#refs/tags/}" >> $GITHUB_ENV
+
+      - name: Checkout release tag
+        run: |
+          git fetch --tags --all --force
+          git checkout tags/${{ env.RELEASE_TAG }}
+
       - name: Set up Go
         uses: actions/setup-go@v5
         with:
@@ -23,9 +64,9 @@ jobs:
       - name: Build production binary
         run: |
           cd src
-          GOOS=linux GOARCH=amd64 go build -trimpath -tags "" -ldflags "-X main.version=${VERSION} -X main.commitHash=${COMMIT_HASH} -X main.buildDate=${BUILD_DATE} -X main.authUrl=$(AUTH_URL) -X main.authClientId=$(AUTH_CLIENT_ID) -X main.authSecret=$(AUTH_CLIENT_SECRET) -X main.sentryDSN=$(SENTRY_DSN)" -o ./build/release/ ./entry...
+          GOOS=linux GOARCH=amd64 GIN_MODE=release go build -trimpath -tags "" -ldflags "-X main.version=${VERSION} -X main.commitHash=${COMMIT_HASH} -X main.buildDate=${BUILD_DATE} -X main.authUrl=$(AUTH_URL) -X main.authClientId=$(AUTH_CLIENT_ID) -X main.authSecret=$(AUTH_CLIENT_SECRET) -X main.sentryDSN=$(SENTRY_DSN)" -o ./build/release/ ./entry...
           mv build/release/entry build/release/ocf-amd64
-          GOOS=linux GOARCH=arm64 go build -trimpath -tags "" -ldflags "-X main.version=${VERSION} -X main.commitHash=${COMMIT_HASH} -X main.buildDate=${BUILD_DATE} -X main.authUrl=$(AUTH_URL) -X main.authClientId=$(AUTH_CLIENT_ID) -X main.authSecret=$(AUTH_CLIENT_SECRET) -X main.sentryDSN=$(SENTRY_DSN)" -o ./build/release/ ./entry...
+          GOOS=linux GOARCH=arm64 GIN_MODE=release go build -trimpath -tags "" -ldflags "-X main.version=${VERSION} -X main.commitHash=${COMMIT_HASH} -X main.buildDate=${BUILD_DATE} -X main.authUrl=$(AUTH_URL) -X main.authClientId=$(AUTH_CLIENT_ID) -X main.authSecret=$(AUTH_CLIENT_SECRET) -X main.sentryDSN=$(SENTRY_DSN)" -o ./build/release/ ./entry...
           mv build/release/entry build/release/ocf-arm64
 
 

local-demo/docker-compose.yml

@@ -8,7 +8,7 @@ networks:
 
 services:
   global-dispatcher:
-    image: ocf-amd64
+    image: ocf-amd64:latest
     build:
       context: ../
       dockerfile: ./local-demo/docker/dispatcher/Dockerfile
@@ -31,10 +31,14 @@ services:
       - 8092:8092
       - 43905:43905
   llm-srevice-1:
-    image: ocf-sql-amd64
+    image: ocf-sql-amd64:latest
     build:
       context: ../
       dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+      args:
+        OCF_IMAGE: ocf-amd64:latest
+    depends_on:
+      - global-dispatcher
     command:
       [
         "start",
@@ -58,10 +62,9 @@ services:
       - 8093:8092
       - 43906:43905
   llm-srevice-2:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -82,10 +85,9 @@ services:
       - 43907:43905
 
   llm-srevice-3:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -106,10 +108,9 @@ services:
       - 43908:43905
 
   llm-srevice-4:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -130,10 +131,9 @@ services:
       - 43910:43905
 
   llm-srevice-5:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -154,10 +154,9 @@ services:
       - 43911:43905
 
   llm-srevice-6:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -178,10 +177,9 @@ services:
       - 43912:43905
 
   llm-srevice-7:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -202,10 +200,9 @@ services:
       - 43913:43905
 
   llm-srevice-8:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -226,10 +223,9 @@ services:
       - 43914:43905
 
   llm-srevice-9:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",
@@ -250,10 +246,9 @@ services:
       - 43915:43905
 
   llm-srevice-0:
-    image: ocf-sql-amd64
-    build:
-      context: ../
-      dockerfile: ./local-demo/docker/llm-serving/Dockerfile
+    image: ocf-sql-amd64:latest
+    depends_on:
+      - llm-srevice-1
     command:
       [
         "start",

local-demo/docker/llm-serving/Dockerfile

@@ -1,14 +1,12 @@
-FROM golang:1.23 AS build
+ARG OCF_IMAGE
+FROM ${OCF_IMAGE} AS build
 
-WORKDIR /app
-
-COPY . /app
-RUN cd /app/src/ && make build-release
 
 FROM python:3.11-slim
 
+
 WORKDIR /app
-COPY --from=build /app/src/build/release/ocf-amd64 .
+COPY --from=build /app/ocf-amd64 .
 
 COPY local-demo/mocked-openai-api/requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt

src/internal/protocol/crdt.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"ocf/internal/common"
+	"strings"
 	"sync"
 	"time"
 
@@ -64,8 +65,12 @@ func GetCRDTStore() (*crdt.Datastore, context.CancelFunc) {
 				p, gerr := GetPeerFromTable(msg.ReceivedFrom.String())
 				if gerr != nil {
 					p = Peer{ID: msg.ReceivedFrom.String()}
+					common.Logger.Infof("Adding peer: [%s] triggered by msg received", msg.ReceivedFrom.String())
+				} else {
+					common.Logger.Infof("Updating peer: [%s] triggered by msg received", msg.ReceivedFrom.String())
 				}
 				p.LastSeen = time.Now().Unix()
+				p.Connected = true
 				if b, merr := json.Marshal(p); merr == nil {
 					UpdateNodeTableHook(ds.NewKey(msg.ReceivedFrom.String()), b)
 				}
@@ -93,11 +98,25 @@ func GetCRDTStore() (*crdt.Datastore, context.CancelFunc) {
 		opts.Logger = common.Logger
 		opts.RebroadcastInterval = 5 * time.Second
 		opts.PutHook = func(k ds.Key, v []byte) {
-			fmt.Printf("Added: [%s] -> %s\n", k, string(v))
 			var peer Peer
 			err := json.Unmarshal(v, &peer)
 			common.ReportError(err, "Error while unmarshalling peer")
-			peer.Connected = true
+			// When a new peer is added to the table it is marked as diconnected by default.
+			// Doing so allows to intercept ghost peers by the verification procedure.
+
+			// Do not update itself
+			host, _ := GetP2PNode(nil)
+			if strings.Trim(k.String(), "/") == host.ID().String() {
+				return
+			}
+			p, err := GetPeerFromTable(strings.Trim(k.String(), "/"))
+			if err != nil {
+				peer.Connected = false
+				common.Logger.Infof("Adding peer: [%s] triggered by p2p hook", strings.Trim(k.String(), "/"))
+			} else {
+				peer.Connected = p.Connected
+				common.Logger.Infof("Updating peer: [%s] triggered by p2p hook", strings.Trim(k.String(), "/"))
+			}
 			value, err := json.Marshal(peer)
 			if err == nil {
 				UpdateNodeTableHook(k, value)
@@ -106,7 +125,7 @@ func GetCRDTStore() (*crdt.Datastore, context.CancelFunc) {
 			}
 		}
 		opts.DeleteHook = func(k ds.Key) {
-			fmt.Printf("Removed: [%s]\n", k)
+			common.Logger.Infof("Removed: [%s] triggered by p2p hook", strings.Trim(k.String(), "/"))
 			DeleteNodeTableHook(k)
 		}
 

src/internal/protocol/host.go

@@ -1,8 +1,11 @@
 package protocol
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	mrand "math/rand"
 	"ocf/internal/common"
@@ -20,10 +23,12 @@ import (
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/pnet"
 	"github.com/libp2p/go-libp2p/core/routing"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
 	"github.com/libp2p/go-libp2p/p2p/security/noise"
 	libp2ptls "github.com/libp2p/go-libp2p/p2p/security/tls"
+	"github.com/libp2p/go-libp2p/p2p/transport/tcp"
 	"github.com/spf13/viper"
 )
 
@@ -32,6 +37,10 @@ var ddht *dualdht.DHT
 var hostOnce sync.Once
 var MyID string
 
+const (
+	Version = "0.0.0-dev.0"
+)
+
 func GetP2PNode(ds datastore.Batching) (host.Host, dualdht.DHT) {
 	hostOnce.Do(func() {
 		ctx := context.Background()
@@ -82,16 +91,29 @@ func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host,
 		return nil, err
 	}
 
+	hash := sha256.Sum256([]byte(Version))
+	keyHex := hex.EncodeToString(hash[:])
+
+	var buf bytes.Buffer
+	buf.WriteString("/key/swarm/psk/1.0.0/\n")
+	buf.WriteString("/base16/\n")
+	buf.WriteString(keyHex + "\n")
+
+	psk, err := pnet.DecodeV1PSK(bytes.NewReader(buf.Bytes()))
+	if err != nil {
+		panic(err)
+	}
+
 	opts := []libp2p.Option{
-		libp2p.DefaultTransports,
+		libp2p.Transport(tcp.NewTCPTransport),
 		libp2p.Identity(priv),
+		libp2p.PrivateNetwork(psk),
 		libp2p.ResourceManager(&network.NullResourceManager{}),
 		// libp2p.ConnectionManager(connmgr),
 		libp2p.NATPortMap(),
 		libp2p.ListenAddrStrings(
 			"/ip4/0.0.0.0/tcp/"+viper.GetString("tcpport"),
 			"/ip4/0.0.0.0/tcp/"+viper.GetString("tcpport")+"/ws",
-			"/ip4/0.0.0.0/udp/"+viper.GetString("udpport")+"/quic",
 		),
 		libp2p.Security(libp2ptls.ID, libp2ptls.New),
 		libp2p.Security(noise.ID, noise.New),
@@ -126,6 +148,9 @@ func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host,
 				p, err := GetPeerFromTable(pid.String())
 				if err != nil {
 					p = Peer{ID: pid.String()}
+					common.Logger.Infof("Adding peer: [%s] triggered by new connection", pid.String())
+				} else {
+					common.Logger.Infof("Updating peer: [%s] triggered by new connection", pid.String())
 				}
 				p.Connected = true
 				p.LastSeen = time.Now().Unix()
@@ -148,6 +173,7 @@ func newHost(ctx context.Context, seed int64, ds datastore.Batching) (host.Host,
 					p = Peer{ID: pid.String()}
 				}
 				p.Connected = false
+				common.Logger.Infof("Removing peer: [%s] triggered by disconnection", pid.String())
 				// keep LastSeen as last known good; do not bump here
 				if b, e := json.Marshal(p); e == nil {
 					UpdateNodeTableHook(datastore.NewKey(pid.String()), b)