maybe this my nginx ideal config, lets go

docker-compose-nginx.yml

@@ -0,0 +1,60 @@
+services:
+  nginx:
+    container_name: balance-zig-pay
+    image: nginx:1.28-alpine
+    volumes:
+      - ./nginx.conf:/etc/nginx/nginx.conf:ro
+    depends_on:
+      - zig1
+      - zig2
+    ports:
+      - "9999:9999"
+    networks:
+      - payment-processor
+    deploy:
+      resources:
+        limits:
+          cpus: "0.50"
+          memory: "100MB"
+
+  zig1: &zig
+    container_name: zig-pay-1
+    build: .
+    environment:
+      - SERVER_ENV=PROD
+      - DEFAULT_HOST=payment-processor-default
+      - DEFAULT_PORT=8080
+      - DEFAULT_POOL=400
+      - FALLBACK_HOST=payment-processor-fallback
+      - FALLBACK_PORT=8080
+      - FALLBACK_POOL=200
+      - PORT=8080
+      - PAYMENT_SUMMARY_EXCHANGE_HOST=zig2
+      - PAYMENT_SUMMARY_EXCHANGE_PORT=6969
+    networks:
+      - payment-processor
+    deploy:
+      resources:
+        limits:
+          cpus: "0.50"
+          memory: "125MB"
+  zig2: 
+    <<: *zig
+    container_name: zig-pay-2
+    environment:
+      - SERVER_ENV=PROD
+      - DEFAULT_HOST=payment-processor-default
+      - DEFAULT_PORT=8080
+      - DEFAULT_POOL=400
+      - FALLBACK_HOST=payment-processor-fallback
+      - FALLBACK_PORT=8080
+      - FALLBACK_POOL=200
+      - PORT=8080
+      - PAYMENT_SUMMARY_EXCHANGE_HOST=zig1
+      - PAYMENT_SUMMARY_EXCHANGE_PORT=6969
+
+networks:
+  backend:
+    driver: bridge
+  payment-processor:
+    external: true

docker-compose.yml

@@ -19,8 +19,8 @@ services:
     deploy:
       resources:
         limits:
-          cpus: "0.20"
-          memory: "50MB"
+          cpus: "1.5"
+          memory: "130MB"
   zig1: &zig
     container_name: zig-pay-1
     build: .
@@ -28,10 +28,10 @@ services:
       - SERVER_ENV=PROD
       - DEFAULT_HOST=payment-processor-default
       - DEFAULT_PORT=8080
-      - DEFAULT_POOL=100
+      - DEFAULT_POOL=700
       - FALLBACK_HOST=payment-processor-fallback
       - FALLBACK_PORT=8080
-      - FALLBACK_POOL=0
+      - FALLBACK_POOL=400
       - PORT=8080
       - PAYMENT_SUMMARY_EXCHANGE_HOST=zig2
       - PAYMENT_SUMMARY_EXCHANGE_PORT=6969
@@ -40,8 +40,8 @@ services:
     deploy:
       resources:
         limits:
-          cpus: "0.20"
-          memory: "50MB"
+          cpus: "0.45"
+          memory: "110MB"
   zig2: 
     <<: *zig
     container_name: zig-pay-2
@@ -49,10 +49,10 @@ services:
       - SERVER_ENV=PROD
       - DEFAULT_HOST=payment-processor-default
       - DEFAULT_PORT=8080
-      - DEFAULT_POOL=100
+      - DEFAULT_POOL=700
       - FALLBACK_HOST=payment-processor-fallback
       - FALLBACK_PORT=8080
-      - FALLBACK_POOL=0
+      - FALLBACK_POOL=400
       - PORT=8080
       - PAYMENT_SUMMARY_EXCHANGE_HOST=zig1
       - PAYMENT_SUMMARY_EXCHANGE_PORT=6969

nginx.conf

@@ -3,11 +3,14 @@ worker_rlimit_nofile 100000;
 
 events {
     use epoll;
-    worker_connections 1024;
+    worker_connections 2000;
 }
+
 http {
     access_log off;
     error_log /dev/null crit;
+    keepalive_requests 10000;
+    keepalive_timeout 600;
 
 
     tcp_nopush on;
@@ -17,12 +20,14 @@ http {
     upstream backend_servers {
         server zig1:8080;
         server zig2:8080;
-        keepalive 500;
+        keepalive 600;
     }
 
     server {
         listen 9999;
         location / {
+            proxy_no_cache 1;
+            proxy_cache_bypass 1;
             proxy_buffering off;
             proxy_http_version 1.1;
             proxy_pass http://backend_servers;

src/context/config.zig

@@ -33,13 +33,13 @@ pub const Config = struct {
         const default_port_env = envMap.get("DEFAULT_PORT");
         const default_port = try std.fmt.parseInt(u16, default_port_env orelse "8001", 10);
         const default_pool_env = envMap.get("DEFAULT_POOL");
-        const default_pool = try std.fmt.parseInt(u16, default_pool_env orelse "20", 10);
+        const default_pool = try std.fmt.parseInt(u16, default_pool_env orelse "200", 10);
 
         const fallback_host = envMap.get("FALLBACK_HOST") orelse "localhost";
         const fallback_port_env = envMap.get("FALLBACK_PORT");
         const fallback_port = try std.fmt.parseInt(u16, fallback_port_env orelse "8002", 10);
         const fallback_pool_env = envMap.get("FALLBACK_POOL");
-        const fallback_pool = try std.fmt.parseInt(u16, fallback_pool_env orelse "10", 10);
+        const fallback_pool = try std.fmt.parseInt(u16, fallback_pool_env orelse "100", 10);
 
         const payment_summary_exchange_host = envMap.get("PAYMENT_SUMMARY_EXCHANGE_HOST");
         const payment_summary_exchange_port_env = envMap.get("PAYMENT_SUMMARY_EXCHANGE_PORT");

src/load_balancer.zig

@@ -12,10 +12,11 @@ const Address = net.Address;
 const Stream = net.Stream;
 
 const BUFFER_SIZE = 1024;
+const MAX_CONNECTION_UPSTREAM = 2050;
 
-var metrics_mutex = std.Thread.Mutex{};
-var metrics_num_requests: u64 = 0;
-var metrics_sum_req_time: u64 = 0;
+//var metrics_mutex = std.Thread.Mutex{};
+//var metrics_num_requests: u64 = 0;
+//var metrics_sum_req_time: u64 = 0;
 
 const UpstreamConnectionState = enum { inactive, available, occupied };
 
@@ -45,7 +46,7 @@ const UpstreamConnection = struct {
 };
 
 pub const UpstreamServer = struct {
-    pool: [300]UpstreamConnection,
+    pool: [MAX_CONNECTION_UPSTREAM]UpstreamConnection,
     address: net.Address,
 
     pub fn init(host: []const u8, port: u16) UpstreamServer {
@@ -67,7 +68,7 @@ pub const UpstreamServer = struct {
 
         std.debug.assert(final_addr != null);
 
-        var connections: [300]UpstreamConnection = undefined;
+        var connections: [MAX_CONNECTION_UPSTREAM]UpstreamConnection = undefined;
 
         for (&connections) |*conn| {
             conn.* = UpstreamConnection.init(final_addr.?);
@@ -128,7 +129,7 @@ pub const LoadBalancer = struct {
 
             upstream.?.occupy();
 
-            var thread = std.Thread.spawn(.{ .stack_size = 1024 * 16 }, handleConnection, .{ self, conn, upstream.? }) catch |err| {
+            var thread = std.Thread.spawn(.{ .stack_size = 1024 * 6 }, handleConnection, .{ self, conn, upstream.? }) catch |err| {
                 log.err("Creating thread error: {}\n", .{err});
                 conn.stream.close();
                 continue;
@@ -151,7 +152,7 @@ pub const LoadBalancer = struct {
 
         buffer_request[0] = 0;
 
-        var timer = std.time.Timer.start() catch return;
+        //var timer = std.time.Timer.start() catch return;
 
         while (true) {
             var req_len: usize = 1;
@@ -174,7 +175,8 @@ pub const LoadBalancer = struct {
                     break;
                 }
             }
-            timer.reset();
+
+            //timer.reset();
 
             upstream.stream.writeAll(buffer_request[0..req_len]) catch |err| {
                 log.err("Error when writing to upstream {}\n", .{err});
@@ -195,17 +197,17 @@ pub const LoadBalancer = struct {
                 return;
             };
 
-            const req_time_ns = timer.lap();
+            //const req_time_ns = timer.lap();
 
-            metrics_mutex.lock();
-            metrics_num_requests += 1;
-            metrics_sum_req_time += req_time_ns;
+            //metrics_mutex.lock();
+            //metrics_num_requests += 1;
+            //metrics_sum_req_time += req_time_ns;
 
-            if (metrics_num_requests % 5000 == 0) {
-                std.debug.print("average requests time ns: {d}\n", .{metrics_sum_req_time / metrics_num_requests});
-            }
+            //if (metrics_num_requests % 5000 == 0) {
+            //    std.debug.print("average requests time ns: {d}\n", .{metrics_sum_req_time / metrics_num_requests});
+            //}
 
-            metrics_mutex.unlock();
+            //metrics_mutex.unlock();
         }
     }
 };

src/main.zig

@@ -26,7 +26,7 @@ pub fn main() !void {
 }
 
 fn startServer() !void {
-    try ctx.intiRepository(100_000);
+    try ctx.intiRepository(1_000_000);
 
     std.debug.print("Starting server...\n", .{});
 

src/payment_integrator.zig

@@ -12,8 +12,8 @@ const DateTime = @import("things").DateTime;
 
 const is_test = @import("builtin").is_test;
 
-const MAX_QUEUE_SIZE = if (is_test) 100 else 100_000;
-const MAX_RETRY = if (is_test) 2 else 10_000_000;
+const MAX_QUEUE_SIZE = if (is_test) 100 else 1_000_000;
+const MAX_RETRY = if (is_test) 2 else 1_000_000;
 
 const PaymentIntegration = struct {
     payment: *payments.Payment,
@@ -53,7 +53,7 @@ pub const PaymentIntegrator = struct {
 
     fn startProcess(self: *PaymentIntegrator) void {
         while (true) {
-            Thread.sleep(50_000_000);
+            Thread.sleep(40_000_000);
 
             self.verifyTailSize();
             self.processPayments();
@@ -79,7 +79,7 @@ pub const PaymentIntegrator = struct {
                     pi.is_integrated = true; // liar
                 } else {
                     pi.payment.requested_at = DateTime.now();
-                    pi.ticket = self.service_pool.dive(pi.getMessage()) catch null;
+                    pi.ticket = self.service_pool.dive(pi.getMessage(), MAX_RETRY - pi.retries) catch null;
                     pi.retries -= 1;
                 }
             } else if (pi.ticket.?.status == .try_again) {
@@ -88,7 +88,8 @@ pub const PaymentIntegrator = struct {
                     pi.payment.integration_status = .not_integrated;
                     pi.is_integrated = true; // liar
                 } else {
-                    pi.ticket = self.service_pool.dive(pi.getMessage()) catch null;
+                    pi.payment.requested_at = DateTime.now();
+                    pi.ticket = self.service_pool.dive(pi.getMessage(), MAX_RETRY - pi.retries) catch null;
                     pi.retries -= 1;
                 }
             } else if (pi.ticket.?.status == .new or pi.ticket.?.status == .pending) {

src/server.zig

@@ -64,7 +64,7 @@ pub const HttpServer = struct {
                 continue;
             };
 
-            var thread = std.Thread.spawn(.{ .stack_size = 1024 * 64 }, handleConnection, .{ self, conn }) catch |err| {
+            var thread = std.Thread.spawn(.{ .stack_size = 1024 * 6 }, handleConnection, .{ self, conn }) catch |err| {
                 log.err("Creating thread error: {}\n", .{err});
                 conn.stream.close();
                 continue;
@@ -126,7 +126,7 @@ pub const HttpServer = struct {
                 .method = http_method,
                 .path = path,
                 .query = query,
-                .body = if (req_len > 70) buffer_request[req_len - 70 .. req_len] else buffer_request[0..0],
+                .body = if (http_method == .POST) buffer_request[findBeginJson(&buffer_request)..req_len] else buffer_request[0..0],
             };
 
             response = Response{
@@ -161,6 +161,16 @@ pub const HttpServer = struct {
         return b;
     }
 
+    fn findBeginJson(buf: []const u8) usize {
+        for (0..buf.len) |i| {
+            if (buf[i] == '{' and (i > 0 and buf[i - 1] == '\n')) {
+                return i;
+            }
+        }
+
+        return if (buf.len == 0) 0 else buf.len - 1;
+    }
+
     pub fn handleReqRes(self: HttpServer, req: *Request, res: *Response) void {
         switch (req.method) {
             .GET => resolveEndpoint(req, res, self.get_endpoints),

src/services/http_service.zig

@@ -241,7 +241,7 @@ pub const HttpService = struct {
             }
 
             self.checkHealth();
-            //log.info("Service {s} health {any} {d} ms\n", .{ self.name, self.health, self.response_time });
+            log.info("Service {s} health {any} {d} ms\n", .{ self.name, self.health, self.response_time });
             Thread.sleep(1_000_000 * 1000);
         }
     }
@@ -305,7 +305,7 @@ pub const HttpService = struct {
 
     fn updateCapacity(self: *HttpService, thread_id: usize) void {
         while (false) {
-            Thread.sleep(500_000_000);
+            Thread.sleep(10_000_000 * 1000);
             var count: usize = 0;
 
             for (self.connections) |conn| {

src/services/service_pool.zig

@@ -27,11 +27,11 @@ pub const ServicePool = struct {
         self.services_len += 1;
     }
 
-    pub fn dive(self: *ServicePool, message: []const u8) ServicePoolError!*ServiceTicket {
-        var skip: [10]bool = .{false} ** 10;
+    pub fn dive(self: *ServicePool, message: []const u8, numberOfRetries: i32) ServicePoolError!*ServiceTicket {
+        //var skip: [10]bool = .{false} ** 10;
 
         for (0..self.services_len) |_| {
-            var best_service = self.findBestService(&skip);
+            var best_service = self.findBestService(numberOfRetries);
 
             if (best_service == null) {
                 return error.NoServiceAvailable;
@@ -49,41 +49,12 @@ pub const ServicePool = struct {
         return error.NoServiceAvailable;
     }
 
-    fn findBestService(self: *ServicePool, skip: []bool) ?*HttpService {
-        var first_service_available_i: ?usize = null;
-        var best_perfomance_i: ?usize = null;
-
-        for (0..self.services_len) |i| {
-            if (skip[i]) {
-                continue;
+    fn findBestService(self: *ServicePool, numberOfRetries: i32) ?*HttpService {
+        if (numberOfRetries > 500) {
+            return self.services[1];
         }
 
-            const service = self.services[i];
-
-            if (service.health == .unavailable or service.capacity >= 90) {
-                continue;
-            }
-
-            if (first_service_available_i == null) {
-                first_service_available_i = i;
-            }
-
-            if (best_perfomance_i == null or self.services[best_perfomance_i.?].response_time > service.response_time) {
-                best_perfomance_i = i;
-            }
-        }
-
-        if (best_perfomance_i != null) {
-            skip[best_perfomance_i.?] = true;
-            return self.services[best_perfomance_i.?];
-        }
-
-        if (first_service_available_i != null) {
-            skip[first_service_available_i.?] = true;
-            return self.services[first_service_available_i.?];
-        }
-
-        return null;
+        return self.services[0];
     }
 };