31 – Communication gRPC Inter-Socles

Vue d’ensemble

Le module gRPC permet aux instances Socle V4 de communiquer entre elles via streaming bidirectionnel. Il offre :

• Sessions : Gestion de sessions multi-participants avec TTL
• Streaming bidirectionnel : Communication temps reel entre Socles
• Pipeline de traitement : Transformation des messages via Janino et JDM
• Fan-out : Routage broadcast ou cible vers les participants
• Pool de connexions : Connexions persistantes vers les peers

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                         Socle A                                  │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐  │
│  │ SessionMgr  │  │  Pipeline   │  │    GrpcServerWorker     │  │
│  │             │  │  Executor   │  │    (port 9400)          │  │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘  │
│         │               │                      │                 │
│         └───────────────┼──────────────────────┘                 │
│                         │                                        │
│              ┌──────────┴──────────┐                            │
│              │   gRPC Services     │                            │
│              │  - SocleComm        │                            │
│              │  - SessionService   │                            │
│              │  - DiscoveryService │                            │
│              └──────────┬──────────┘                            │
└─────────────────────────┼───────────────────────────────────────┘
                          │ gRPC/HTTP2
                          ▼
┌─────────────────────────────────────────────────────────────────┐
│                         Socle B                                  │
│              ┌──────────────────────┐                           │
│              │  PeerConnectionPool  │                           │
│              └──────────────────────┘                           │
└─────────────────────────────────────────────────────────────────┘

Configuration

application.yml

socle:
  grpc:
    # Activation du module
    enabled: ${GRPC_ENABLED:false}

    # Port du serveur gRPC
    port: ${GRPC_PORT:9400}

    # Identification du Socle
    socle-id: ${SOCLE_ID:${socle.app_name}}
    socle-version: ${socle.version}

    # Limites serveur
    max-inbound-message-size: ${GRPC_MAX_MESSAGE_SIZE:4194304}  # 4MB
    max-concurrent-calls-per-connection: ${GRPC_MAX_CONCURRENT_CALLS:100}

    # Sessions
    session:
      ttl-seconds: ${GRPC_SESSION_TTL:1800}           # 30 min
      max-participants: ${GRPC_MAX_PARTICIPANTS:100}
      persist-to-tech-db: ${GRPC_PERSIST_SESSIONS:true}
      cache-in-redis: ${GRPC_CACHE_REDIS:true}

    # Pipeline de traitement
    pipeline:
      enabled: ${GRPC_PIPELINE_ENABLED:true}
      config-cache-ttl-seconds: ${GRPC_PIPELINE_CACHE_TTL:300}

    # Connexions peer
    peer:
      max-channels-per-peer: ${GRPC_PEER_MAX_CHANNELS:4}
      connection-timeout-ms: ${GRPC_PEER_CONNECT_TIMEOUT:5000}
      idle-timeout-seconds: ${GRPC_PEER_IDLE_TIMEOUT:300}
      keep-alive-enabled: ${GRPC_PEER_KEEPALIVE:true}
      keep-alive-time-seconds: ${GRPC_PEER_KEEPALIVE_TIME:30}
      keep-alive-timeout-seconds: ${GRPC_PEER_KEEPALIVE_TIMEOUT:10}

Variables d’environnement

Services gRPC

DiscoveryService

Service de decouverte et health check.

service DiscoveryService {
    rpc GetCapabilities(CapabilitiesRequest) returns (CapabilitiesResponse);
    rpc Ping(PingRequest) returns (PingResponse);
}

Test avec grpcurl :

# Ping
grpcurl -plaintext localhost:9400 socle.DiscoveryService/Ping

# Capabilities
grpcurl -plaintext localhost:9400 socle.DiscoveryService/GetCapabilities

SessionService

Gestion du cycle de vie des sessions.

service SessionService {
    rpc CreateSession(CreateSessionRequest) returns (SessionInfo);
    rpc JoinSession(JoinSessionRequest) returns (JoinSessionResponse);
    rpc LeaveSession(LeaveSessionRequest) returns (LeaveSessionResponse);
    rpc GetSession(GetSessionRequest) returns (SessionInfo);
    rpc CloseSession(CloseSessionRequest) returns (CloseSessionResponse);
}

Exemples :

# Creer une session
grpcurl -plaintext -d '{
  "session_type": "chat",
  "owner_id": "user1",
  "ttl_seconds": 3600
}' localhost:9400 socle.SessionService/CreateSession

# Joindre une session
grpcurl -plaintext -d '{
  "session_id": "uuid-de-la-session",
  "participant_id": "user2",
  "display_name": "User 2"
}' localhost:9400 socle.SessionService/JoinSession

# Obtenir info session
grpcurl -plaintext -d '{
  "session_id": "uuid-de-la-session"
}' localhost:9400 socle.SessionService/GetSession

SocleComm

Streaming bidirectionnel pour l’echange de messages.

service SocleComm {
    rpc Exchange(stream SessionMessage) returns (stream SessionMessage);
}

Format des messages :

message SessionMessage {
    string session_id = 1;
    string sender_id = 2;
    MessageKind kind = 3;        // JOIN, LEAVE, DATA, REQUEST, RESPONSE, etc.
    repeated string target_ids = 4;  // Vide = broadcast
    string correlation_id = 5;
    int64 timestamp = 6;
    string payload = 7;          // JSON
    map<string, string> headers = 8;
}

Sessions

Cycle de vie

CREATE ──► ACTIVE ──► CLOSING ──► CLOSED
              │
              └──► EXPIRED (TTL depasse)

Stockage

Les sessions sont stockees dans :

1. Cache memoire : Acces rapide, stream observers
2. Redis (si KvBus en mode redis) : Partage entre instances
3. TechDB : Audit et persistance

Tables TechDB

-- Sessions (audit)
CREATE TABLE grpc_sessions (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    session_id VARCHAR(36) NOT NULL UNIQUE,
    session_type VARCHAR(100) NOT NULL,
    owner_id VARCHAR(100) NOT NULL,
    status VARCHAR(20) DEFAULT 'ACTIVE',
    datas CLOB
);

-- Configuration pipeline par type de session
CREATE TABLE grpc_pipeline_configs (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    session_type VARCHAR(100) NOT NULL UNIQUE,
    janino_pre VARCHAR(255),
    jdm_rules VARCHAR(255),
    janino_post VARCHAR(255),
    default_targets VARCHAR(50) DEFAULT 'broadcast',
    enabled BOOLEAN DEFAULT TRUE,
    datas CLOB
);

Pipeline de traitement

Le pipeline permet de transformer les messages et determiner leur routage.

Etapes

Message entrant
      │
      ▼
┌─────────────────┐
│  1. Janino PRE  │  Transformation du payload
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  2. JDM Rules   │  Determination des cibles (routing)
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  3. Janino POST │  Transformation finale
└────────┬────────┘
         │
         ▼
    Routage
   /        \
Broadcast   Cible

Configuration du pipeline

Inserez dans grpc_pipeline_configs :

INSERT INTO grpc_pipeline_configs
(session_type, janino_pre, jdm_rules, janino_post, default_targets, enabled)
VALUES
('chat', NULL, 'chat_routing', NULL, 'broadcast', TRUE);

Exemple script Janino PRE

// repository/scripts/java/grpc/chat_filter.java
public class ChatFilter {
    public Object execute(Map<String, Object> context) {
        Map<String, Object> payload = (Map<String, Object>) context.get("payload");
        String senderId = (String) context.get("senderId");

        // Ajouter metadata
        payload.put("processed_at", System.currentTimeMillis());
        payload.put("sender", senderId);

        return payload;
    }
}

Exemple modele JDM pour routing

{
  "name": "chat_routing",
  "nodes": [
    {
      "id": "input",
      "type": "inputNode",
      "content": {
        "fields": [
          {"field": "payload.target", "type": "string"},
          {"field": "payload.type", "type": "string"}
        ]
      }
    },
    {
      "id": "decision",
      "type": "decisionTableNode",
      "content": {
        "hitPolicy": "first",
        "inputs": [
          {"field": "payload.type"}
        ],
        "outputs": [
          {"field": "targets", "type": "string"},
          {"field": "broadcast", "type": "boolean"}
        ],
        "rules": [
          {"_input": ["private"], "_output": ["${payload.target}", false]},
          {"_input": ["broadcast"], "_output": ["", true]},
          {"_input": ["*"], "_output": ["", true]}
        ]
      }
    }
  ]
}

Connexions Peer

Ajouter un peer programmatiquement

@Autowired
private PeerConnectionPool peerPool;

public void connectToPeer() {
    // Ajouter et connecter
    boolean connected = peerPool.addPeer("socle-b", "192.168.1.100", 9400);

    if (connected) {
        // Ouvrir un stream
        peerPool.openStream("socle-b", message -> {
            // Handler pour messages entrants
            System.out.println("Received: " + message.getPayload());
        });

        // Envoyer un message
        SessionMessage msg = SessionMessage.newBuilder()
            .setSessionId("...")
            .setSenderId("local-participant")
            .setKind(MessageKind.DATA)
            .setPayload("{\"text\":\"Hello\"}")
            .build();

        peerPool.sendToPeer("socle-b", msg);
    }
}

Broadcast vers tous les peers

SessionMessage msg = SessionMessage.newBuilder()
    .setSessionId(sessionId)
    .setSenderId(myId)
    .setKind(MessageKind.DATA)
    .setPayload(jsonPayload)
    .build();

int sent = peerPool.broadcast(msg, null);  // null = inclure tous

Worker gRPC

Priorites

Statistiques

@Autowired
private GrpcServerWorker grpcWorker;

Map<String, Object> stats = grpcWorker.getStats();
// {
//   "running": true,
//   "port": 9400,
//   "sessions": { "cached_sessions": 5, "active_streams": 12 },
//   "messaging": { "messages_received": 1234, "messages_sent": 5678 },
//   "peers": { "connected_peers": 2, "total_messages_sent": 100 }
// }

Types de sessions supportes

Securite

TLS (a venir)

Le support TLS sera ajoute dans une version future :

socle:
  grpc:
    tls:
      enabled: true
      cert-path: /path/to/server.crt
      key-path: /path/to/server.key
      ca-path: /path/to/ca.crt  # Pour mTLS

Authentification

L’authentification peut etre implementee via :

• Metadata gRPC (tokens dans headers)
• Intercepteurs personnalises
• Integration avec le module Auth existant

Monitoring

Metriques exposees

Les metriques sont disponibles via l’endpoint Prometheus /actuator/prometheus :

• grpc_sessions_active : Nombre de sessions actives
• grpc_messages_received_total : Total messages recus
• grpc_messages_sent_total : Total messages envoyes
• grpc_peers_connected : Nombre de peers connectes

Logs

[grpc_server] Started on port 9400
[grpc_session] Session created: abc-123 (type=chat, owner=user1)
[grpc_session] Participant joined: user2 -> abc-123 (total: 2)
[grpc_comm] Message received: session=abc-123, sender=user1, kind=DATA
[grpc_pipeline] Executed for session abc-123 (targets=broadcast)

Troubleshooting

Le serveur ne demarre pas

1. Verifiez que le port 9400 n’est pas utilise
2. Verifiez GRPC_ENABLED=true
3. Consultez les logs pour les erreurs d’initialisation

Sessions expirent trop vite

Augmentez le TTL :

socle:
  grpc:
    session:
      ttl-seconds: 7200  # 2 heures

Messages non delivres

1. Verifiez que le destinataire a un stream actif
2. Verifiez les logs du pipeline pour les erreurs
3. Verifiez la configuration du routing dans grpc_pipeline_configs

Erreurs de connexion peer

1. Verifiez la connectivite reseau
2. Verifiez que le peer a gRPC active
3. Augmentez le timeout de connexion :

socle:
  grpc:
    peer:
      connection-timeout-ms: 10000

Structure des fichiers

src/main/java/eu/lmvi/socle/grpc/
├── GrpcConfiguration.java        # Configuration Spring
├── GrpcServerWorker.java         # Worker principal
├── session/
│   ├── Session.java              # Record session
│   ├── Participant.java          # Record participant
│   ├── SessionStatus.java        # Enum ACTIVE/CLOSING/CLOSED/EXPIRED
│   ├── SessionManager.java       # Gestion sessions + streams
│   └── SessionRedisRepository.java  # Persistance Redis
├── pipeline/
│   ├── PipelineConfig.java       # Config pipeline
│   ├── PipelineResult.java       # Resultat pipeline
│   └── PipelineExecutor.java     # Execution Janino + JDM
├── peer/
│   ├── PeerConnection.java       # Connexion unique
│   └── PeerConnectionPool.java   # Pool avec reconnexion
└── service/
    ├── SocleCommService.java     # Streaming bidirectionnel
    ├── SessionServiceImpl.java   # CRUD sessions
    └── DiscoveryServiceImpl.java # Discovery + Ping

src/main/proto/
└── socle_comm.proto              # Definition Protocol Buffers

03 – Guide de Démarrage Rapide

Version : 4.0.0 Date : 2025-01-25 Temps estimé : 5 minutes

1. Prérequis

• JDK 21+ installé
• Maven 3.9+ installé
• Git (optionnel)

Vérification :

java -version   # java version "21.x.x"
mvn -version    # Maven 3.9.x

2. Installation

Option A : Cloner le projet

git clone <repo>/socle-v004.git
cd socle-v004

Option B : Créer un nouveau projet

mvn archetype:generate \
  -DgroupId=com.mycompany \
  -DartifactId=my-socle-app \
  -DarchetypeArtifactId=maven-archetype-quickstart \
  -DinteractiveMode=false

cd my-socle-app
# Ajouter la dépendance socle-v004 dans pom.xml

3. Configuration minimale

3.1 pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>eu.lmvi</groupId>
    <artifactId>socle-v004</artifactId>
    <version>4.0.0</version>
    <packaging>jar</packaging>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.1</version>
    </parent>

    <properties>
        <java.version>21</java.version>
    </properties>

    <dependencies>
        <!-- Spring Boot Web (sans Logback) -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <!-- Log4j2 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-log4j2</artifactId>
        </dependency>

        <!-- LMAX Disruptor (AsyncLoggers) -->
        <dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>4.0.0</version>
        </dependency>

        <!-- H2 Database -->
        <dependency>
            <groupId>com.h2database</groupId>
            <artifactId>h2</artifactId>
            <version>2.2.224</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

3.2 application.yml

spring:
  application:
    name: my-socle-app

socle:
  app_name: ${APP_NAME:my-socle-app}
  version: 4.0.0
  env_name: ${ENV_NAME:DEV}
  region: ${REGION:local}

  # HTTP
  http_port: ${HTTP_PORT:8080}

  # KV Bus
  kv_impl: in_memory

  # Supervisor
  supervisor_enabled: true

  # H2 TechDB
  techdb:
    enabled: true
    url: jdbc:h2:file:./data/techdb;MODE=PostgreSQL
    console:
      enabled: true
      path: /h2-console

logging:
  config: classpath:log4j2.xml

3.3 log4j2.xml

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
    <Appenders>
        <Console name="Console" target="SYSTEM_OUT">
            <PatternLayout pattern="%d{ISO8601} %-5level [%thread] %logger{36} - %msg%n"/>
        </Console>
    </Appenders>
    <Loggers>
        <Root level="INFO">
            <AppenderRef ref="Console"/>
        </Root>
    </Loggers>
</Configuration>

4. Créer un Worker simple

package com.mycompany.worker;

import eu.lmvi.socle.worker.Worker;
import eu.lmvi.socle.supervisor.Supervisor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class HelloWorker implements Worker {

    private static final Logger log = LoggerFactory.getLogger(HelloWorker.class);

    @Autowired
    private Supervisor supervisor;  // Injection du Supervisor

    private int counter = 0;
    private volatile boolean running = false;

    @Override
    public String getName() {
        return "hello-worker";
    }

    @Override
    public void initialize() {
        log.info("HelloWorker initializing...");
    }

    @Override
    public void start() {
        running = true;
        log.info("HelloWorker started!");
    }

    @Override
    public void doWork() {
        if (!running) return;  // Vérifier le flag d'arrêt

        try {
            counter++;
            log.info("Hello from worker! Count: {}", counter);

            // IMPORTANT: Envoyer un heartbeat avec métriques
            supervisor.heartbeat(getName(), getStats());

        } catch (Exception e) {
            log.error("Erreur dans doWork", e);
            // NE PAS propager l'exception
        }
    }

    @Override
    public void stop() {
        running = false;
        log.info("HelloWorker stopped. Final count: {}", counter);
    }

    @Override
    public boolean isHealthy() {
        return running;
    }

    @Override
    public Map<String, Object> getStats() {
        return Map.of(
            "counter", counter,
            "running", running
        );
    }

    @Override
    public long getCycleIntervalMs() {
        return 5000; // doWork() toutes les 5 secondes
    }
}

5. Build et Run

# Build
mvn clean package -DskipTests

# Run
java -jar target/socle-v004-4.0.0.jar

# Ou avec variables d'environnement
APP_NAME=my-app ENV_NAME=PROD java -jar target/socle-v004-4.0.0.jar

6. Vérification

Health check

curl http://localhost:8080/health

Réponse attendue :

{
  "status": "UP",
  "app": "my-socle-app",
  "version": "4.0.0"
}

H2 Console (dev)

Ouvrir : http://localhost:8080/h2-console

• JDBC URL : jdbc:h2:file:./data/techdb
• User : socle
• Password : socle

Logs

Vous devriez voir :

INFO  [main] MainOrchestratorProcess - MOP démarré
INFO  [main] HelloWorker - HelloWorker started!
INFO  [scheduler-1] HelloWorker - Hello from worker! Count: 1
INFO  [scheduler-1] HelloWorker - Hello from worker! Count: 2

StatusDashboard (supervision)

Ouvrir : http://localhost:9374/

Ce dashboard affiche en temps réel :

• L’état de chaque worker (RUNNING, DEGRADED, UNHEALTHY)
• Les métriques transmises par les heartbeats
• L’historique des changements d’état

7. Prochaines étapes

8. Troubleshooting

Port déjà utilisé

# Changer le port
HTTP_PORT=9090 java -jar target/socle-v004-4.0.0.jar

H2 Console inaccessible

Vérifier que socle.techdb.console.enabled: true dans application.yml.

Logs non visibles

Vérifier que log4j2.xml existe dans src/main/resources/.

23 – Client Authentification JWT (Nouveauté V4)

Version : 4.0.0 Date : 2025-12-09

1. Introduction

Le SocleAuthClient est un client d’authentification JWT intégré au Socle V4 pour communiquer avec les services centraux (LogHub, Registry, etc.).

Pattern d’authentification

┌─────────────────┐                    ┌─────────────────┐
│   Application   │                    │  Auth Server    │
│   Socle V4      │                    │  (central)      │
└────────┬────────┘                    └────────┬────────┘
         │                                      │
         │  1. POST /auth/login                 │
         │     {sourceName, apiKey}             │
         │─────────────────────────────────────►│
         │                                      │
         │  2. {accessToken, refreshToken}      │
         │◄─────────────────────────────────────│
         │                                      │
         │  3. Requêtes avec Bearer token       │
         │  Authorization: Bearer <accessToken> │
         │─────────────────────────────────────►│ Services
         │                                      │
         │  4. POST /auth/refresh (auto)        │
         │     {refreshToken}                   │
         │─────────────────────────────────────►│
         │                                      │
         │  5. {accessToken (new)}              │
         │◄─────────────────────────────────────│

2. Configuration

2.1 application.yml

socle:
  auth:
    enabled: ${AUTH_ENABLED:false}
    server-url: ${AUTH_SERVER_URL:https://auth.lmvi.org}
    source-name: ${SOURCE_NAME:${socle.app_name}}
    api-key: ${API_KEY:}
    access-token-buffer-seconds: 60
    connect-timeout-ms: 10000
    read-timeout-ms: 30000

2.2 Variables d’environnement

3. Interface SocleAuthClient

package eu.lmvi.socle.client.auth;

/**
 * Client d'authentification Socle V4
 */
public interface SocleAuthClient {

    /**
     * Login initial avec API Key
     * @return Tokens d'accès et de refresh
     * @throws AuthenticationException si échec
     */
    AuthTokens login() throws AuthenticationException;

    /**
     * Refresh du token d'accès
     * @param refreshToken Token de refresh
     * @return Nouveaux tokens
     * @throws AuthenticationException si échec
     */
    AuthTokens refresh(String refreshToken) throws AuthenticationException;

    /**
     * Obtenir un token d'accès valide (avec refresh auto si nécessaire)
     * @return Token d'accès valide
     * @throws AuthenticationException si échec
     */
    String getValidAccessToken() throws AuthenticationException;

    /**
     * Vérifie si le client est authentifié
     * @return true si un token valide existe
     */
    boolean isAuthenticated();

    /**
     * Invalide les tokens courants
     */
    void logout();
}

4. DTOs

4.1 AuthTokens

package eu.lmvi.socle.client.auth;

public record AuthTokens(
    String accessToken,
    String refreshToken,
    Instant accessTokenExpiry,
    Instant refreshTokenExpiry
) {
    public boolean isAccessTokenExpired() {
        return Instant.now().isAfter(accessTokenExpiry);
    }

    public boolean isAccessTokenExpiringSoon(int bufferSeconds) {
        return Instant.now().plusSeconds(bufferSeconds).isAfter(accessTokenExpiry);
    }

    public boolean isRefreshTokenExpired() {
        return Instant.now().isAfter(refreshTokenExpiry);
    }
}

4.2 LoginRequest / LoginResponse

// Request
public record LoginRequest(
    String sourceName,
    String apiKey
) {}

// Response
public record LoginResponse(
    String accessToken,
    String refreshToken,
    int expiresIn,        // secondes
    int refreshExpiresIn  // secondes
) {}

5. Implémentation AuthTokenManager

package eu.lmvi.socle.client.auth;

@Component
@ConditionalOnProperty(name = "socle.auth.enabled", havingValue = "true")
public class AuthTokenManager implements SocleAuthClient {

    private static final Logger log = LoggerFactory.getLogger(AuthTokenManager.class);

    private final SocleConfiguration config;
    private final OkHttpClient httpClient;
    private final ObjectMapper objectMapper;

    private volatile AuthTokens currentTokens;
    private final ReentrantLock refreshLock = new ReentrantLock();

    public AuthTokenManager(SocleConfiguration config) {
        this.config = config;
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());

        this.httpClient = new OkHttpClient.Builder()
            .connectTimeout(config.getAuthConnectTimeoutMs(), TimeUnit.MILLISECONDS)
            .readTimeout(config.getAuthReadTimeoutMs(), TimeUnit.MILLISECONDS)
            .build();
    }

    @Override
    public AuthTokens login() throws AuthenticationException {
        log.info("Login to auth server: {}", config.getAuthServerUrl());

        LoginRequest request = new LoginRequest(
            config.getSourceName(),
            config.getApiKey()
        );

        try {
            String json = objectMapper.writeValueAsString(request);

            Request httpRequest = new Request.Builder()
                .url(config.getAuthServerUrl() + "/api/v1/auth/login")
                .post(RequestBody.create(json, MediaType.parse("application/json")))
                .build();

            try (Response response = httpClient.newCall(httpRequest).execute()) {
                if (!response.isSuccessful()) {
                    throw new AuthenticationException("Login failed: " + response.code());
                }

                LoginResponse loginResponse = objectMapper.readValue(
                    response.body().string(),
                    LoginResponse.class
                );

                currentTokens = new AuthTokens(
                    loginResponse.accessToken(),
                    loginResponse.refreshToken(),
                    Instant.now().plusSeconds(loginResponse.expiresIn()),
                    Instant.now().plusSeconds(loginResponse.refreshExpiresIn())
                );

                log.info("Login successful, token expires in {} seconds", loginResponse.expiresIn());
                return currentTokens;
            }
        } catch (IOException e) {
            throw new AuthenticationException("Login failed", e);
        }
    }

    @Override
    public AuthTokens refresh(String refreshToken) throws AuthenticationException {
        log.debug("Refreshing access token");

        RefreshRequest request = new RefreshRequest(refreshToken);

        try {
            String json = objectMapper.writeValueAsString(request);

            Request httpRequest = new Request.Builder()
                .url(config.getAuthServerUrl() + "/api/v1/auth/refresh")
                .post(RequestBody.create(json, MediaType.parse("application/json")))
                .build();

            try (Response response = httpClient.newCall(httpRequest).execute()) {
                if (!response.isSuccessful()) {
                    // Refresh failed, need to re-login
                    log.warn("Refresh failed, attempting re-login");
                    return login();
                }

                RefreshResponse refreshResponse = objectMapper.readValue(
                    response.body().string(),
                    RefreshResponse.class
                );

                currentTokens = new AuthTokens(
                    refreshResponse.accessToken(),
                    currentTokens.refreshToken(),  // Keep same refresh token
                    Instant.now().plusSeconds(refreshResponse.expiresIn()),
                    currentTokens.refreshTokenExpiry()
                );

                log.debug("Token refreshed, new expiry in {} seconds", refreshResponse.expiresIn());
                return currentTokens;
            }
        } catch (IOException e) {
            throw new AuthenticationException("Refresh failed", e);
        }
    }

    @Override
    public String getValidAccessToken() throws AuthenticationException {
        // First time - login
        if (currentTokens == null) {
            login();
            return currentTokens.accessToken();
        }

        // Refresh token expired - need full re-login
        if (currentTokens.isRefreshTokenExpired()) {
            log.info("Refresh token expired, re-login required");
            login();
            return currentTokens.accessToken();
        }

        // Access token expiring soon - refresh
        int bufferSeconds = config.getAccessTokenBufferSeconds();
        if (currentTokens.isAccessTokenExpiringSoon(bufferSeconds)) {
            refreshLock.lock();
            try {
                // Double-check after acquiring lock
                if (currentTokens.isAccessTokenExpiringSoon(bufferSeconds)) {
                    refresh(currentTokens.refreshToken());
                }
            } finally {
                refreshLock.unlock();
            }
        }

        return currentTokens.accessToken();
    }

    @Override
    public boolean isAuthenticated() {
        return currentTokens != null && !currentTokens.isAccessTokenExpired();
    }

    @Override
    public void logout() {
        currentTokens = null;
        log.info("Logged out");
    }
}

6. Utilisation

6.1 Injection

@Service
public class MonService {

    @Autowired(required = false)
    private SocleAuthClient authClient;

    public void callSecuredApi() throws Exception {
        if (authClient == null || !authClient.isAuthenticated()) {
            throw new IllegalStateException("Auth not configured");
        }

        String token = authClient.getValidAccessToken();

        // Utiliser le token
        Request request = new Request.Builder()
            .url("https://api.mycompany.com/secured")
            .header("Authorization", "Bearer " + token)
            .build();

        // ...
    }
}

6.2 Intégration avec LogForwarder

// Dans HttpLogTransport
public class HttpLogTransport implements LogTransport {

    private final SocleAuthClient authClient;

    @Override
    public void send(List<LogEntry> entries) throws Exception {
        String token = authClient.getValidAccessToken();

        Request request = new Request.Builder()
            .url(logHubUrl)
            .header("Authorization", "Bearer " + token)
            .post(RequestBody.create(toJson(entries), JSON))
            .build();

        // ...
    }
}

6.3 Intégration avec MOP

// Dans MainOrchestratorProcess.start()
if (authClient != null && config.isAuthEnabled()) {
    log.info("[step:auth] Login auprès du serveur d'auth");
    try {
        authClient.login();
    } catch (AuthenticationException e) {
        log.error("Auth failed, continuing without auth", e);
    }
}

7. Gestion des erreurs

7.1 AuthenticationException

public class AuthenticationException extends Exception {
    public AuthenticationException(String message) {
        super(message);
    }

    public AuthenticationException(String message, Throwable cause) {
        super(message, cause);
    }
}

7.2 Retry automatique

Le AuthTokenManager gère automatiquement :

• Le refresh avant expiration
• Le re-login si le refresh échoue
• Le re-login si le refresh token expire

7.3 Fallback sans auth

if (authClient == null) {
    log.warn("Auth not configured, proceeding without authentication");
    // Continuer sans auth (pour dev local)
}

8. Sécurité

8.1 Stockage de l’API Key

# NE PAS mettre dans le code
# Utiliser des variables d'environnement
export API_KEY="xxx-secret-key"

# Ou un gestionnaire de secrets
# Kubernetes Secret, AWS Secrets Manager, etc.

8.2 Tokens en mémoire

Les tokens sont stockés en mémoire uniquement :

• Jamais persistés sur disque
• Invalidés au restart
• Thread-safe

8.3 HTTPS obligatoire

socle:
  auth:
    server-url: https://auth.mycompany.com  # HTTPS obligatoire

9. Monitoring

9.1 Métriques

// Exposées via /metrics
socle_auth_login_total          # Nombre de logins
socle_auth_login_errors_total   # Erreurs de login
socle_auth_refresh_total        # Nombre de refresh
socle_auth_token_expiry_seconds # Temps avant expiration

9.2 Logs

INFO  - Login to auth server: https://auth.lmvi.org
INFO  - Login successful, token expires in 900 seconds
DEBUG - Refreshing access token
DEBUG - Token refreshed, new expiry in 900 seconds
WARN  - Refresh failed, attempting re-login

10. Troubleshooting

Connection refused

AuthenticationException: Login failed: Connection refused

Vérifier :

• AUTH_SERVER_URL est correct
• Le serveur d’auth est accessible
• Les ports sont ouverts

Invalid API Key

AuthenticationException: Login failed: 401

Vérifier :

• API_KEY est correct
• SOURCE_NAME est enregistré côté serveur

Token expired

Si les tokens expirent trop vite :

• Vérifier l’horloge système (NTP)
• Augmenter access-token-buffer-seconds

11. Références

• RFC 7519 – JWT
• OAuth 2.0
• 10-SECURITY

10 – Security

Version : 4.0.0 Date : 2025-12-09

1. Introduction

Le Socle V4 intègre plusieurs mécanismes de sécurité :

• Authentification Admin API (Basic Auth)
• Client JWT pour services externes (nouveauté V4)
• Filtrage des endpoints sensibles
• Gestion sécurisée des secrets

2. Authentification Admin API

2.1 Configuration

socle:
  admin:
    enabled: true
    auth:
      enabled: ${ADMIN_AUTH_ENABLED:false}
      username: ${ADMIN_USERNAME:admin}
      password: ${ADMIN_PASSWORD:}

2.2 AdminAuthFilter

package eu.lmvi.socle.security;

@Component
@Order(1)
public class AdminAuthFilter implements Filter {

    private final SocleConfiguration config;

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
            throws IOException, ServletException {

        HttpServletRequest httpRequest = (HttpServletRequest) request;
        HttpServletResponse httpResponse = (HttpServletResponse) response;

        // Skip si auth désactivée
        if (!config.getAdmin().getAuth().isEnabled()) {
            chain.doFilter(request, response);
            return;
        }

        // Skip les endpoints publics
        String path = httpRequest.getRequestURI();
        if (isPublicEndpoint(path)) {
            chain.doFilter(request, response);
            return;
        }

        // Vérifier l'authentification pour /admin/*
        if (path.startsWith("/admin")) {
            String authHeader = httpRequest.getHeader("Authorization");

            if (!isValidAuth(authHeader)) {
                httpResponse.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
                httpResponse.setHeader("WWW-Authenticate", "Basic realm=\"Admin API\"");
                return;
            }
        }

        chain.doFilter(request, response);
    }

    private boolean isPublicEndpoint(String path) {
        return path.equals("/admin/health") || path.equals("/admin/health/live");
    }

    private boolean isValidAuth(String authHeader) {
        if (authHeader == null || !authHeader.startsWith("Basic ")) {
            return false;
        }

        String base64Credentials = authHeader.substring("Basic ".length());
        String credentials = new String(Base64.getDecoder().decode(base64Credentials));
        String[] parts = credentials.split(":", 2);

        if (parts.length != 2) {
            return false;
        }

        return parts[0].equals(config.getAdmin().getAuth().getUsername())
            && parts[1].equals(config.getAdmin().getAuth().getPassword());
    }
}

2.3 Utilisation

# Sans auth
curl http://localhost:8080/admin/health

# Avec auth
curl -u admin:secret http://localhost:8080/admin/workers

3. JWT Auth Client (Nouveauté V4)

3.1 Principe

Le SocleAuthClient permet aux applications Socle de s’authentifier auprès de services centraux (LogHub, Registry, etc.).

Application Socle  ──────►  Auth Server  ──────►  Services sécurisés
      │                          │                      │
      │  1. Login (API Key)      │                      │
      │─────────────────────────►│                      │
      │                          │                      │
      │  2. JWT tokens           │                      │
      │◄─────────────────────────│                      │
      │                          │                      │
      │  3. Requêtes avec Bearer │                      │
      │──────────────────────────────────────────────────►

3.2 Configuration

socle:
  auth:
    enabled: ${AUTH_ENABLED:false}
    server-url: ${AUTH_SERVER_URL:https://auth.mycompany.com}
    source-name: ${SOURCE_NAME:${socle.app_name}}
    api-key: ${API_KEY:}
    access-token-buffer-seconds: 60

3.3 Utilisation

@Service
public class SecuredApiClient {

    @Autowired(required = false)
    private SocleAuthClient authClient;

    public void callSecuredApi() {
        if (authClient == null || !authClient.isAuthenticated()) {
            throw new SecurityException("Authentication required");
        }

        String token = authClient.getValidAccessToken();

        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create("https://api.mycompany.com/data"))
            .header("Authorization", "Bearer " + token)
            .build();

        // ...
    }
}

Voir 23-AUTH-CLIENT pour la documentation complète.

4. Gestion des secrets

4.1 Variables d’environnement

# JAMAIS dans le code ou les fichiers committes
export ADMIN_PASSWORD="super-secret-password"
export API_KEY="my-api-key"
export REDIS_PASSWORD="redis-password"
export TECHDB_PASSWORD="techdb-password"

4.2 Docker Secrets

# docker-compose.yml
services:
  app:
    image: socle-v4:latest
    secrets:
      - admin_password
      - api_key
    environment:
      - ADMIN_PASSWORD_FILE=/run/secrets/admin_password
      - API_KEY_FILE=/run/secrets/api_key

secrets:
  admin_password:
    file: ./secrets/admin_password.txt
  api_key:
    file: ./secrets/api_key.txt

4.3 Kubernetes Secrets

apiVersion: v1
kind: Secret
metadata:
  name: socle-secrets
type: Opaque
stringData:
  ADMIN_PASSWORD: "super-secret"
  API_KEY: "my-api-key"
---
apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
        - name: app
          envFrom:
            - secretRef:
                name: socle-secrets

4.4 Configuration sécurisée

@Configuration
public class SecretsConfiguration {

    @Value("${ADMIN_PASSWORD:#{null}}")
    private String adminPassword;

    @Value("${ADMIN_PASSWORD_FILE:#{null}}")
    private String adminPasswordFile;

    @PostConstruct
    public void loadSecrets() {
        // Charger depuis fichier si spécifié
        if (adminPasswordFile != null) {
            try {
                adminPassword = Files.readString(Path.of(adminPasswordFile)).trim();
            } catch (IOException e) {
                throw new RuntimeException("Failed to load secret from file", e);
            }
        }
    }

    public String getAdminPassword() {
        return adminPassword;
    }
}

5. CORS Configuration

@Configuration
public class CorsConfiguration implements WebMvcConfigurer {

    @Value("${socle.security.cors.allowed-origins:*}")
    private String allowedOrigins;

    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/api/**")
            .allowedOrigins(allowedOrigins.split(","))
            .allowedMethods("GET", "POST", "PUT", "DELETE")
            .allowedHeaders("*")
            .exposedHeaders("X-Total-Count")
            .allowCredentials(true)
            .maxAge(3600);
    }
}

6. Rate Limiting

6.1 Implémentation simple

@Component
public class RateLimitFilter implements Filter {

    private final KvBus kvBus;
    private final int maxRequests = 100;
    private final Duration window = Duration.ofMinutes(1);

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
            throws IOException, ServletException {

        HttpServletRequest httpRequest = (HttpServletRequest) request;
        HttpServletResponse httpResponse = (HttpServletResponse) response;

        String clientId = getClientId(httpRequest);
        String key = "ratelimit:" + clientId + ":" + Instant.now().truncatedTo(ChronoUnit.MINUTES);

        long count = kvBus.increment(key);
        if (count == 1) {
            kvBus.setTtl(key, window);
        }

        if (count > maxRequests) {
            httpResponse.setStatus(429);
            httpResponse.setHeader("Retry-After", "60");
            httpResponse.getWriter().write("Rate limit exceeded");
            return;
        }

        httpResponse.setHeader("X-RateLimit-Limit", String.valueOf(maxRequests));
        httpResponse.setHeader("X-RateLimit-Remaining", String.valueOf(maxRequests - count));

        chain.doFilter(request, response);
    }

    private String getClientId(HttpServletRequest request) {
        String apiKey = request.getHeader("X-API-Key");
        if (apiKey != null) {
            return apiKey;
        }
        return request.getRemoteAddr();
    }
}

7. Input Validation

7.1 Validation des DTOs

public record CreateOrderRequest(
    @NotNull @Size(min = 1, max = 100)
    String customerId,

    @NotEmpty
    List<@Valid OrderItem> items,

    @Email
    String notificationEmail
) {}

public record OrderItem(
    @NotNull @Size(min = 1, max = 50)
    String productId,

    @Min(1) @Max(1000)
    int quantity
) {}

7.2 Controller avec validation

@RestController
@RequestMapping("/api/orders")
public class OrderController {

    @PostMapping
    public ResponseEntity<Order> createOrder(@Valid @RequestBody CreateOrderRequest request) {
        // request est validé automatiquement
        return ResponseEntity.ok(orderService.create(request));
    }
}

7.3 Sanitization

public class InputSanitizer {

    private static final Pattern SAFE_STRING = Pattern.compile("^[a-zA-Z0-9-_]+$");

    public static String sanitizeId(String input) {
        if (input == null) return null;
        if (!SAFE_STRING.matcher(input).matches()) {
            throw new IllegalArgumentException("Invalid ID format");
        }
        return input;
    }

    public static String sanitizeForLog(String input) {
        if (input == null) return null;
        return input.replaceAll("[\n\r\t]", "_");
    }
}

8. Logging sécurisé

8.1 Ne pas logger les secrets

// MAUVAIS
log.info("Connecting with password: {}", password);
log.info("API Key: {}", apiKey);
log.info("Request: {}", requestWithSensitiveData);

// BON
log.info("Connecting to database");
log.info("Using API Key: {}...", apiKey.substring(0, 4));
log.info("Request received for user: {}", sanitizeForLog(userId));

8.2 Pattern pour masquer les données sensibles

public class SecureLogger {

    private static final Set<String> SENSITIVE_FIELDS = Set.of(
        "password", "apiKey", "token", "secret", "credential"
    );

    public static String maskSensitiveData(Map<String, Object> data) {
        Map<String, Object> masked = new HashMap<>();
        for (Map.Entry<String, Object> entry : data.entrySet()) {
            if (SENSITIVE_FIELDS.contains(entry.getKey().toLowerCase())) {
                masked.put(entry.getKey(), "***MASKED***");
            } else {
                masked.put(entry.getKey(), entry.getValue());
            }
        }
        return masked.toString();
    }
}

9. Headers de sécurité

@Component
public class SecurityHeadersFilter implements Filter {

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
            throws IOException, ServletException {

        HttpServletResponse httpResponse = (HttpServletResponse) response;

        // Prevent clickjacking
        httpResponse.setHeader("X-Frame-Options", "DENY");

        // XSS protection
        httpResponse.setHeader("X-Content-Type-Options", "nosniff");
        httpResponse.setHeader("X-XSS-Protection", "1; mode=block");

        // HSTS (si HTTPS)
        httpResponse.setHeader("Strict-Transport-Security", "max-age=31536000; includeSubDomains");

        // Content Security Policy
        httpResponse.setHeader("Content-Security-Policy", "default-src 'self'");

        chain.doFilter(request, response);
    }
}

10. Audit logging

@Aspect
@Component
public class AuditAspect {

    private static final Logger auditLog = LoggerFactory.getLogger("AUDIT");

    @Around("@annotation(Audited)")
    public Object audit(ProceedingJoinPoint joinPoint) throws Throwable {
        String method = joinPoint.getSignature().getName();
        String user = getCurrentUser();
        Instant start = Instant.now();

        try {
            Object result = joinPoint.proceed();
            auditLog.info("SUCCESS | user={} | method={} | duration={}ms",
                user, method, Duration.between(start, Instant.now()).toMillis());
            return result;
        } catch (Exception e) {
            auditLog.warn("FAILURE | user={} | method={} | error={}",
                user, method, e.getMessage());
            throw e;
        }
    }

    private String getCurrentUser() {
        // Récupérer l'utilisateur du contexte
        return "system";
    }
}

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface Audited {}

11. Checklist de sécurité

Configuration

• [ ] ADMIN_AUTH_ENABLED=true en production
• [ ] Mots de passe forts et uniques
• [ ] Secrets via variables d’environnement ou secret manager
• [ ] HTTPS activé
• [ ] CORS configuré correctement

Code

• [ ] Validation de tous les inputs
• [ ] Pas de secrets dans les logs
• [ ] Headers de sécurité activés
• [ ] Rate limiting en place
• [ ] Audit logging activé

Infrastructure

• [ ] Firewall configuré
• [ ] Ports non nécessaires fermés
• [ ] H2 Console désactivée en production
• [ ] Accès admin restreint

12. Références

• 23-AUTH-CLIENT – Client JWT
• 13-TLS-HTTPS – Configuration TLS
• 14-ADMIN-API – API Admin
• OWASP Top 10

Plan de Documentation – Socle V4

Version : 4.0.0 Date : 2025-01-25

Structure de la documentation

Nouveautés V4

Les documents 21 à 30 sont spécifiques au Socle V4 :

• 21-H2-TECHDB : Base embarquée H2 pour état technique
• 22-LOG4J2-LOGFORWARDER : Migration Logback → Log4j2 + centralisation logs
• 23-AUTH-CLIENT : Client JWT pour services centraux
• 24-WORKER-REGISTRY : Auto-enregistrement des workers
• 25-MIGRATION-V3-V4 : Guide de migration depuis V3
• 26-GRAALVM-JAVASCRIPT : GraalVM CE 21 et GraalJS pour exécution JavaScript
• 27-STATUS-DASHBOARD : Dashboard HTML de supervision temps réel sur port 9374
• 29-JANINO : Compilation dynamique de scripts Java
• 30-EVENTBUS-WORKERS : Workers orientés événements
• 31-GRPC-INTER-SOCLES : Communication gRPC bidirectionnelle entre Socles

Guide Méthodologique

Un guide méthodologique complet est disponible pour aider les développeurs à implémenter leurs solutions :

• GUIDE-METHODOLOGIQUE : Bonnes pratiques, patterns, cas d’usage

Ce document répond à la question : « Je dois implémenter X, comment je fais ? »

Conventions

• Tous les fichiers sont en Markdown
• Les exemples de code sont en Java 21
• Les configurations sont en YAML
• Les commandes sont pour Linux/macOS (adaptables Windows)

11 – Resilience (Circuit Breaker & Retry)

Version : 4.0.0 Date : 2025-12-09

1. Introduction

Le Socle V4 intègre des patterns de résilience pour gérer les défaillances des systèmes externes :

• Retry : Réessayer les opérations échouées
• Circuit Breaker : Protéger contre les cascades de défaillances

2. Retry Pattern

2.1 Configuration

socle:
  resilience:
    retry:
      max-attempts: ${RETRY_MAX_ATTEMPTS:3}
      initial-delay-ms: ${RETRY_INITIAL_DELAY_MS:1000}
      max-delay-ms: ${RETRY_MAX_DELAY_MS:30000}
      multiplier: ${RETRY_MULTIPLIER:2.0}

2.2 Interface RetryTemplate

package eu.lmvi.socle.resilience;

public interface RetryTemplate {

    /**
     * Exécute une opération avec retry
     */
    <T> T execute(Supplier<T> operation) throws RetryExhaustedException;

    /**
     * Exécute une opération avec retry et fallback
     */
    <T> T executeWithFallback(Supplier<T> operation, Supplier<T> fallback);

    /**
     * Exécute une opération void avec retry
     */
    void executeVoid(Runnable operation) throws RetryExhaustedException;
}

2.3 Implémentation

package eu.lmvi.socle.resilience;

@Component
public class ExponentialBackoffRetryTemplate implements RetryTemplate {

    private static final Logger log = LoggerFactory.getLogger(ExponentialBackoffRetryTemplate.class);

    private final int maxAttempts;
    private final long initialDelayMs;
    private final long maxDelayMs;
    private final double multiplier;

    public ExponentialBackoffRetryTemplate(SocleConfiguration config) {
        this.maxAttempts = config.getResilience().getRetry().getMaxAttempts();
        this.initialDelayMs = config.getResilience().getRetry().getInitialDelayMs();
        this.maxDelayMs = config.getResilience().getRetry().getMaxDelayMs();
        this.multiplier = config.getResilience().getRetry().getMultiplier();
    }

    @Override
    public <T> T execute(Supplier<T> operation) throws RetryExhaustedException {
        Exception lastException = null;
        long delay = initialDelayMs;

        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            try {
                return operation.get();
            } catch (Exception e) {
                lastException = e;
                log.warn("Attempt {}/{} failed: {}", attempt, maxAttempts, e.getMessage());

                if (attempt < maxAttempts) {
                    sleep(delay);
                    delay = Math.min((long) (delay * multiplier), maxDelayMs);
                }
            }
        }

        throw new RetryExhaustedException(
            "Operation failed after " + maxAttempts + " attempts",
            lastException
        );
    }

    @Override
    public <T> T executeWithFallback(Supplier<T> operation, Supplier<T> fallback) {
        try {
            return execute(operation);
        } catch (RetryExhaustedException e) {
            log.warn("All retries exhausted, using fallback");
            return fallback.get();
        }
    }

    @Override
    public void executeVoid(Runnable operation) throws RetryExhaustedException {
        execute(() -> {
            operation.run();
            return null;
        });
    }

    private void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Retry interrupted", e);
        }
    }
}

2.4 Utilisation

@Service
public class ExternalApiService {

    @Autowired
    private RetryTemplate retryTemplate;

    public Data fetchData(String id) {
        return retryTemplate.execute(() -> {
            // Appel qui peut échouer
            return httpClient.get("/data/" + id);
        });
    }

    public Data fetchDataWithFallback(String id) {
        return retryTemplate.executeWithFallback(
            () -> httpClient.get("/data/" + id),
            () -> getCachedData(id)  // Fallback vers le cache
        );
    }
}

3. Circuit Breaker Pattern

3.1 États

        succès
     ┌─────────────────┐
     │                 │
     ▼                 │
┌─────────┐      ┌─────┴─────┐      ┌─────────┐
│  CLOSED │─────►│   OPEN    │─────►│HALF_OPEN│
└────┬────┘      └───────────┘      └────┬────┘
     │  échecs        │ timeout          │
     │  threshold     │                  │
     │                │                  │
     └────────────────┴──────────────────┘
              retour succès

• CLOSED : Fonctionnement normal, les requêtes passent
• OPEN : Circuit ouvert, les requêtes échouent immédiatement
• HALF_OPEN : Test de reprise, quelques requêtes passent

3.2 Configuration

socle:
  resilience:
    circuit-breaker:
      failure-threshold: ${CB_FAILURE_THRESHOLD:5}
      success-threshold: ${CB_SUCCESS_THRESHOLD:3}
      timeout-ms: ${CB_TIMEOUT_MS:60000}
      half-open-requests: ${CB_HALF_OPEN_REQUESTS:3}

3.3 Interface CircuitBreaker

package eu.lmvi.socle.resilience;

public interface CircuitBreaker {

    /**
     * Nom du circuit
     */
    String getName();

    /**
     * État actuel
     */
    CircuitState getState();

    /**
     * Exécute une opération protégée
     */
    <T> T execute(Supplier<T> operation) throws CircuitBreakerOpenException;

    /**
     * Exécute avec fallback
     */
    <T> T executeWithFallback(Supplier<T> operation, Supplier<T> fallback);

    /**
     * Force l'ouverture
     */
    void forceOpen();

    /**
     * Force la fermeture
     */
    void forceClose();

    /**
     * Reset les compteurs
     */
    void reset();

    /**
     * Métriques
     */
    CircuitBreakerMetrics getMetrics();
}

public enum CircuitState {
    CLOSED,
    OPEN,
    HALF_OPEN
}

3.4 Implémentation

package eu.lmvi.socle.resilience;

public class DefaultCircuitBreaker implements CircuitBreaker {

    private static final Logger log = LoggerFactory.getLogger(DefaultCircuitBreaker.class);

    private final String name;
    private final int failureThreshold;
    private final int successThreshold;
    private final long timeoutMs;
    private final int halfOpenRequests;

    private volatile CircuitState state = CircuitState.CLOSED;
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicInteger successCount = new AtomicInteger(0);
    private final AtomicInteger halfOpenCount = new AtomicInteger(0);
    private volatile Instant lastFailureTime;
    private final ReentrantLock lock = new ReentrantLock();

    @Override
    public <T> T execute(Supplier<T> operation) throws CircuitBreakerOpenException {
        if (!allowRequest()) {
            throw new CircuitBreakerOpenException(name);
        }

        try {
            T result = operation.get();
            onSuccess();
            return result;
        } catch (Exception e) {
            onFailure();
            throw e;
        }
    }

    @Override
    public <T> T executeWithFallback(Supplier<T> operation, Supplier<T> fallback) {
        try {
            return execute(operation);
        } catch (CircuitBreakerOpenException e) {
            log.debug("[{}] Circuit open, using fallback", name);
            return fallback.get();
        } catch (Exception e) {
            log.warn("[{}] Operation failed, using fallback", name, e);
            return fallback.get();
        }
    }

    private boolean allowRequest() {
        switch (state) {
            case CLOSED:
                return true;

            case OPEN:
                // Vérifier si le timeout est passé
                if (lastFailureTime != null &&
                    Duration.between(lastFailureTime, Instant.now()).toMillis() > timeoutMs) {
                    transitionTo(CircuitState.HALF_OPEN);
                    return true;
                }
                return false;

            case HALF_OPEN:
                // Limiter les requêtes en half-open
                return halfOpenCount.incrementAndGet() <= halfOpenRequests;

            default:
                return false;
        }
    }

    private void onSuccess() {
        lock.lock();
        try {
            switch (state) {
                case CLOSED:
                    failureCount.set(0);
                    break;

                case HALF_OPEN:
                    if (successCount.incrementAndGet() >= successThreshold) {
                        transitionTo(CircuitState.CLOSED);
                    }
                    break;
            }
        } finally {
            lock.unlock();
        }
    }

    private void onFailure() {
        lock.lock();
        try {
            lastFailureTime = Instant.now();

            switch (state) {
                case CLOSED:
                    if (failureCount.incrementAndGet() >= failureThreshold) {
                        transitionTo(CircuitState.OPEN);
                    }
                    break;

                case HALF_OPEN:
                    transitionTo(CircuitState.OPEN);
                    break;
            }
        } finally {
            lock.unlock();
        }
    }

    private void transitionTo(CircuitState newState) {
        if (state != newState) {
            log.info("[{}] Circuit state: {} -> {}", name, state, newState);
            state = newState;
            failureCount.set(0);
            successCount.set(0);
            halfOpenCount.set(0);
        }
    }

    @Override
    public CircuitState getState() {
        return state;
    }

    @Override
    public String getName() {
        return name;
    }

    @Override
    public void forceOpen() {
        transitionTo(CircuitState.OPEN);
    }

    @Override
    public void forceClose() {
        transitionTo(CircuitState.CLOSED);
    }

    @Override
    public void reset() {
        lock.lock();
        try {
            state = CircuitState.CLOSED;
            failureCount.set(0);
            successCount.set(0);
            halfOpenCount.set(0);
            lastFailureTime = null;
        } finally {
            lock.unlock();
        }
    }
}

3.5 CircuitBreakerRegistry

@Component
public class CircuitBreakerRegistry {

    private final ConcurrentHashMap<String, CircuitBreaker> circuits = new ConcurrentHashMap<>();
    private final SocleConfiguration config;

    public CircuitBreaker getOrCreate(String name) {
        return circuits.computeIfAbsent(name, this::createCircuitBreaker);
    }

    public CircuitBreaker get(String name) {
        return circuits.get(name);
    }

    public Map<String, CircuitState> getAllStates() {
        return circuits.entrySet().stream()
            .collect(Collectors.toMap(
                Map.Entry::getKey,
                e -> e.getValue().getState()
            ));
    }

    private CircuitBreaker createCircuitBreaker(String name) {
        return new DefaultCircuitBreaker(
            name,
            config.getResilience().getCircuitBreaker().getFailureThreshold(),
            config.getResilience().getCircuitBreaker().getSuccessThreshold(),
            config.getResilience().getCircuitBreaker().getTimeoutMs(),
            config.getResilience().getCircuitBreaker().getHalfOpenRequests()
        );
    }
}

3.6 Utilisation

@Service
public class PaymentService {

    @Autowired
    private CircuitBreakerRegistry cbRegistry;

    public PaymentResult processPayment(Payment payment) {
        CircuitBreaker cb = cbRegistry.getOrCreate("payment-gateway");

        return cb.executeWithFallback(
            () -> paymentGateway.process(payment),
            () -> {
                // Fallback: mettre en queue pour traitement ultérieur
                paymentQueue.enqueue(payment);
                return PaymentResult.pending("Queued for later processing");
            }
        );
    }
}

4. Combinaison Retry + Circuit Breaker

@Service
public class ResilientApiClient {

    @Autowired
    private RetryTemplate retryTemplate;

    @Autowired
    private CircuitBreakerRegistry cbRegistry;

    public Data fetchData(String endpoint) {
        CircuitBreaker cb = cbRegistry.getOrCreate("api-" + endpoint);

        return cb.executeWithFallback(
            () -> retryTemplate.execute(() -> httpClient.get(endpoint)),
            () -> getCachedData(endpoint)
        );
    }
}

Ordre d’exécution

1. CircuitBreaker vérifie si le circuit est ouvert
   → Si OPEN: fallback immédiat
   → Si CLOSED/HALF_OPEN: continue

2. RetryTemplate essaie l'opération
   → Retry avec backoff exponentiel
   → Si tous les retries échouent: exception

3. CircuitBreaker compte l'échec
   → Si seuil atteint: passe en OPEN

4. Fallback si échec

5. Annotations (optionnel)

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Resilient {
    String circuitBreaker() default "";
    int maxRetries() default 3;
    long retryDelay() default 1000;
    Class<? extends Throwable>[] retryOn() default {Exception.class};
}

@Aspect
@Component
public class ResilienceAspect {

    @Around("@annotation(resilient)")
    public Object handleResilience(ProceedingJoinPoint pjp, Resilient resilient) throws Throwable {
        String cbName = resilient.circuitBreaker();
        if (cbName.isEmpty()) {
            cbName = pjp.getSignature().getDeclaringTypeName() + "." + pjp.getSignature().getName();
        }

        CircuitBreaker cb = cbRegistry.getOrCreate(cbName);

        return cb.execute(() -> {
            return retryTemplate.execute(() -> {
                try {
                    return pjp.proceed();
                } catch (Throwable t) {
                    throw new RuntimeException(t);
                }
            });
        });
    }
}

Utilisation

@Service
public class UserService {

    @Resilient(circuitBreaker = "user-api", maxRetries = 5)
    public User getUser(String id) {
        return userApiClient.fetchUser(id);
    }
}

6. Bulkhead Pattern

Le pattern Bulkhead limite le nombre d’appels concurrents pour éviter l’épuisement des ressources.

@Component
public class BulkheadRegistry {

    private final ConcurrentHashMap<String, Semaphore> bulkheads = new ConcurrentHashMap<>();

    public <T> T execute(String name, int maxConcurrent, Supplier<T> operation)
            throws BulkheadFullException {
        Semaphore semaphore = bulkheads.computeIfAbsent(name,
            k -> new Semaphore(maxConcurrent));

        if (!semaphore.tryAcquire()) {
            throw new BulkheadFullException(name);
        }

        try {
            return operation.get();
        } finally {
            semaphore.release();
        }
    }
}

// Utilisation
@Service
public class ApiService {

    @Autowired
    private BulkheadRegistry bulkheadRegistry;

    public Data callExternalApi() {
        return bulkheadRegistry.execute("external-api", 10, () -> {
            // Max 10 appels concurrents
            return httpClient.get("/api/data");
        });
    }
}

7. Timeout Pattern

@Component
public class TimeoutTemplate {

    private final ScheduledExecutorService scheduler =
        Executors.newScheduledThreadPool(4);

    public <T> T executeWithTimeout(Supplier<T> operation, Duration timeout)
            throws TimeoutException {

        CompletableFuture<T> future = CompletableFuture.supplyAsync(operation);

        try {
            return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
        } catch (java.util.concurrent.TimeoutException e) {
            future.cancel(true);
            throw new TimeoutException("Operation timed out after " + timeout);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

8. API Admin

@RestController
@RequestMapping("/admin/resilience")
public class ResilienceController {

    @Autowired
    private CircuitBreakerRegistry cbRegistry;

    @GetMapping("/circuits")
    public Map<String, CircuitState> getCircuits() {
        return cbRegistry.getAllStates();
    }

    @PostMapping("/circuits/{name}/reset")
    public void resetCircuit(@PathVariable String name) {
        CircuitBreaker cb = cbRegistry.get(name);
        if (cb != null) {
            cb.reset();
        }
    }

    @PostMapping("/circuits/{name}/open")
    public void openCircuit(@PathVariable String name) {
        CircuitBreaker cb = cbRegistry.get(name);
        if (cb != null) {
            cb.forceOpen();
        }
    }

    @PostMapping("/circuits/{name}/close")
    public void closeCircuit(@PathVariable String name) {
        CircuitBreaker cb = cbRegistry.get(name);
        if (cb != null) {
            cb.forceClose();
        }
    }
}

9. Métriques

@Component
public class ResilienceMetrics {

    private final MeterRegistry registry;

    public void recordRetry(String operation, int attempt, boolean success) {
        Counter.builder("socle_retry_attempts")
            .tag("operation", operation)
            .tag("attempt", String.valueOf(attempt))
            .tag("success", String.valueOf(success))
            .register(registry)
            .increment();
    }

    public void recordCircuitBreakerState(String name, CircuitState state) {
        Gauge.builder("socle_circuit_breaker_state", () ->
            state == CircuitState.CLOSED ? 0 :
            state == CircuitState.HALF_OPEN ? 1 : 2)
            .tag("name", name)
            .register(registry);
    }
}

10. Bonnes pratiques

DO

• Utiliser le circuit breaker pour les appels réseau externes
• Configurer des timeouts appropriés
• Toujours prévoir un fallback
• Monitorer l’état des circuits
• Logger les transitions d’état

DON’T

• Ne pas utiliser le retry pour les erreurs non récupérables (400, 401)
• Ne pas configurer des seuils trop bas (faux positifs)
• Ne pas oublier les timeouts (risque de blocage)
• Ne pas ignorer les métriques

11. Références

• 05-WORKERS – Workers
• 09-PIPELINE – Pipeline
• Martin Fowler – Circuit Breaker
• Resilience4j

28 – Standard de Creation de Table H2

Version : 4.0.1 Date : 2026-01-13 Package : eu.lmvi.socle.techdb

Introduction

Ce document definit le standard de creation des tables dans la base technique H2 du Socle V004. Ce standard garantit la coherence, la tracabilite et la compatibilite PostgreSQL/H2.

Structure de Base

Toutes les tables TechDB doivent suivre cette structure:

Regles

Identite

• Utiliser GENERATED BY DEFAULT AS IDENTITY, pas AUTO_INCREMENT (MySQL)
• Pas de sequence explicite
• La colonne x_id est toujours la cle primaire

-- Correct (SQL standard / H2 / PostgreSQL)
x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY

-- Incorrect (MySQL uniquement)
id BIGINT AUTO_INCREMENT PRIMARY KEY

Timestamps

• x_dateCreated: Toujours NOT NULL DEFAULT CURRENT_TIMESTAMP
• x_dateChanged: NULL a l’insertion, mis a jour par l’application
• Utiliser TIMESTAMP WITH TIME ZONE pour la compatibilite

x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
x_dateChanged TIMESTAMP WITH TIME ZONE,

Triggers

• Aucun trigger dans H2
• L’audit, l’historisation et la mise a jour de x_dateChanged sont geres par l’application

Cle Primaire

• Si une cle existante est presente (ex: worker_name), la conserver comme UNIQUE
• x_id reste la PK technique

x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
worker_name VARCHAR(255) NOT NULL UNIQUE,

Conventions de Nommage

Permissions

• Droits geres au niveau utilisateur H2
• Proprietaire = utilisateur createur (socle par defaut)

Contraintes H2 vs PostgreSQL

Exemple DDL Complet

CREATE TABLE IF NOT EXISTS techdb_example (
    -- Champs techniques (standard)
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,

    -- Champs metier specifiques
    example_key VARCHAR(255) NOT NULL UNIQUE,
    status VARCHAR(50) NOT NULL,
    counter INT DEFAULT 0,
    last_activity TIMESTAMP WITH TIME ZONE,

    -- Donnees JSON (toujours en dernier)
    datas CLOB
);

-- Index recommandes
CREATE INDEX IF NOT EXISTS idx_example_key ON techdb_example(example_key);
CREATE INDEX IF NOT EXISTS idx_example_status ON techdb_example(status);
CREATE INDEX IF NOT EXISTS idx_example_created ON techdb_example(x_dateCreated);

Tables TechDB du Socle

Le Socle V004 definit 5 tables techniques:

techdb_offsets

Stockage des offsets de consommation (Kafka, NATS, etc.)

CREATE TABLE IF NOT EXISTS techdb_offsets (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    offset_key VARCHAR(255) NOT NULL UNIQUE,
    topic VARCHAR(255) NOT NULL,
    partition_id INT DEFAULT 0,
    offset_value BIGINT NOT NULL,
    consumer_group VARCHAR(255),
    datas CLOB
);

techdb_worker_state

Etat persistant des Workers

CREATE TABLE IF NOT EXISTS techdb_worker_state (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    worker_name VARCHAR(255) NOT NULL UNIQUE,
    state VARCHAR(50) NOT NULL,
    last_run_at TIMESTAMP WITH TIME ZONE,
    next_run_at TIMESTAMP WITH TIME ZONE,
    error_count INT DEFAULT 0,
    last_error CLOB,
    datas CLOB
);

techdb_events

Evenements techniques

CREATE TABLE IF NOT EXISTS techdb_events (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    event_type VARCHAR(100) NOT NULL,
    source VARCHAR(255) NOT NULL,
    processed BOOLEAN DEFAULT FALSE,
    processed_at TIMESTAMP WITH TIME ZONE,
    datas CLOB
);

techdb_log_buffer

Buffer de logs pour LogForwarder

CREATE TABLE IF NOT EXISTS techdb_log_buffer (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    log_level VARCHAR(20) NOT NULL,
    logger_name VARCHAR(255),
    message CLOB NOT NULL,
    thread_name VARCHAR(255),
    log_timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
    forwarded BOOLEAN DEFAULT FALSE,
    forwarded_at TIMESTAMP WITH TIME ZONE,
    datas CLOB
);

techdb_kv

Stockage cle-valeur generique

CREATE TABLE IF NOT EXISTS techdb_kv (
    x_id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    x_dateCreated TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
    x_dateChanged TIMESTAMP WITH TIME ZONE,
    x_sub VARCHAR(255),
    x_partition VARCHAR(30),
    x_comment CLOB,
    kv_key VARCHAR(512) NOT NULL UNIQUE,
    value_type VARCHAR(50) DEFAULT 'string',
    expires_at TIMESTAMP WITH TIME ZONE,
    datas CLOB
);

Bonnes Pratiques

DO

• Toujours utiliser le format standard x_ pour les champs techniques
• Mettre datas en derniere colonne
• Creer des index sur les colonnes frequemment requetees
• Utiliser TIMESTAMP WITH TIME ZONE pour tous les timestamps
• Documenter le contenu JSON attendu dans datas

DON’T

• Ne pas utiliser AUTO_INCREMENT (syntaxe MySQL)
• Ne pas creer de triggers (gestion applicative)
• Ne pas stocker de BLOBs volumineux (utiliser stockage externe)
• Ne pas utiliser JSONB (non supporte par H2)
• Ne pas omettre les index sur les colonnes de recherche

Migration depuis l’ancien format

Si vous avez des tables existantes avec l’ancien format:

-- 1. Sauvegarder les donnees
CREATE TABLE techdb_events_backup AS SELECT * FROM techdb_events;

-- 2. Supprimer l'ancienne table
DROP TABLE techdb_events;

-- 3. Recreer avec le nouveau format
-- (voir DDL ci-dessus)

-- 4. Migrer les donnees
INSERT INTO techdb_events (event_type, source, processed, processed_at, datas)
SELECT event_type, source, processed, processed_at, payload
FROM techdb_events_backup;

-- 5. Supprimer la sauvegarde
DROP TABLE techdb_events_backup;

Voir aussi

• 21-H2-TECHDB.md – Documentation TechDB complete
• 05-WORKERS.md – Workers TechDB

H2 = dev/tests/outillage – Structure compatible PostgreSQL/H2

Socle V004 – Standard TechDB