fix(sync): pool multiplexed consumer connections, real queue topics, ssh carries sync settings; regenerate openapi.json

Modified app/githost/src/main.rs +2 −1 View app/githost/src/main.rs

@@ -29,7 +29,8 @@

2929 .with_kv(Some(Arc::new(state.redis.clone())))

3030 .with_queue(Some(sync_queue))

3131 .with_sync_settings(cfg.git.sync.clone());

32 let ssh_svc = git::ssh::service::GitSshService::new(smart_svc.clone(), cfg.git.ssh.clone());

32 let ssh_svc = git::ssh::service::GitSshService::new(smart_svc.clone(), cfg.git.ssh.clone())

33 .with_sync_settings(cfg.git.sync.clone());

3334 let smart_router = git::smart::handler::router(smart_svc);

3435

3536 let http_addr = SocketAddr::new(cfg.githost.http_bind_addr(), cfg.githost.http_port());

Modified lib/config/src/git.rs +2 −2 View lib/config/src/git.rs

@@ -107,8 +107,8 @@

107107 lock_ttl_secs: 120,

108108 max_commits_per_sync: 5_000,

109109 concurrency: 4,

110 queue_topic: "gitdata:{tasks}".to_string(),

111 dlq_topic: "gitdata:{dlq}".to_string(),

110 queue_topic: "gitdata:tasks".to_string(),

111 dlq_topic: "gitdata:dlq".to_string(),

112112 max_lock_retries: 5,

113113 max_languages: 20,

114114 language_timeout_secs: 30,

Modified lib/git/src/sync/queue.rs +56 −7 View lib/git/src/sync/queue.rs

@@ -1,6 +1,7 @@

11use export::{AppRedis, RedisConnection};

2use redis::aio::MultiplexedConnection;

23

3use crate::sync::error::SyncResult;

4use crate::sync::error::{SyncError, SyncResult};

45

56macro_rules! dispatch {

67 ($conn:expr, |$c:ident| $body:expr) => {

@@ -16,12 +17,29 @@

1617#[derive(Clone)]

1718pub struct SyncQueue {

1819 redis: AppRedis,

20 consumer_conns: Vec<MultiplexedConnection>,

1921}

2022

2123impl SyncQueue {

2224 #[must_use]

2325 pub fn new(redis: AppRedis) -> Self {

24 Self { redis }

26 Self {

27 redis,

28 consumer_conns: Vec::new(),

29 }

30 }

31

32 pub async fn with_consumer_conns(redis: AppRedis, count: usize) -> SyncResult<Self> {

33 let count = count.max(1);

34 let mut consumer_conns = Vec::with_capacity(count);

35 for _ in 0..count {

36 let conn = open_multiplexed(&redis).await?;

37 consumer_conns.push(conn);

38 }

39 Ok(Self {

40 redis,

41 consumer_conns,

42 })

2543 }

2644

2745 #[must_use]

@@ -29,21 +47,39 @@

2947 &self.redis

3048 }

3149

32 pub async fn consume(&self, key: &str, timeout_secs: u64) -> SyncResult<Option<Vec<u8>>> {

33 let mut conn = self.redis.connection_clone();

50 #[must_use]

51 pub fn consumer_conns(&self) -> &[MultiplexedConnection] {

52 &self.consumer_conns

53 }

54

55 pub async fn consume(

56 &self,

57 key: &str,

58 timeout_secs: u64,

59 slot: usize,

60 ) -> SyncResult<Option<Vec<u8>>> {

3461 let res: Option<(String, Vec<u8>)> = if timeout_secs == 0 {

62 let mut conn = self.redis.connection_clone();

3563 dispatch!(conn, |c| {

3664 redis::cmd("RPOP").arg(key).query_async(c).await

37 })

65 })?

66 } else if let Some(conn) = self.consumer_conns.get(slot).cloned() {

67 let mut conn = conn;

68 redis::cmd("BRPOP")

69 .arg(key)

70 .arg(timeout_secs)

71 .query_async(&mut conn)

72 .await?

3873 } else {

74 let mut conn = self.redis.connection_clone();

3975 dispatch!(conn, |c| {

4076 redis::cmd("BRPOP")

4177 .arg(key)

4278 .arg(timeout_secs)

4379 .query_async(c)

4480 .await

45 })

46 }?;

81 })?

82 };

4783 Ok(res.map(|(_, v)| v))

4884 }

4985

@@ -90,3 +126,16 @@

90126 Ok(())

91127 }

92128}

129

130async fn open_multiplexed(redis: &AppRedis) -> SyncResult<MultiplexedConnection> {

131 let url = redis.config().url().to_string();

132 let client = redis::Client::open(url.as_str())

133 .map_err(|e| SyncError::Internal(format!("open redis client for consumer: {e}")))?;

134 let config = redis::AsyncConnectionConfig::new()

135 .set_response_timeout(None)

136 .set_connection_timeout(Some(redis.config().connection_timeout()));

137 client

138 .get_multiplexed_async_connection_with_config(&config)

139 .await

140 .map_err(|e| SyncError::Internal(format!("open multiplexed consumer connection: {e}")))

141}

Modified openapi.json View openapi.json

Diff too large to display.