nx_ha_replication.nx source

1// nx_ha_replication.nx -- CAP-HA-REPLICATION: append-only log-shipping replication (WAL-style) to kill the DATA 2// SPOF (the single NAS). The primary appends records; rep_ship copies new bytes to a standby/backup so it holds a 3// BYTE-EXACT prefix of the primary. Usable TODAY as backup/DR (single-node), and the sync target when a standby host 4// exists. Append-only = history sacred (never rewrites). Deterministic; rep_in_sync proves byte-identity, rep_lag 5// measures how far behind the replica is. license_tier: ORIGINAL 6import "nx_syscalls.nx" 7const K_MAGIC_1048576: i64 = 1048576 8 9// end position (bytes) of a log; 0 if absent/empty. 10func rep_position(path: *u8) -> i64 { 11 let fd: i64 = sys_openat_rd(path); if fd < 0 { return 0 } 12 let sz: i64 = sys_lseek(fd, 0, 2); sys_close(fd) // whence 2 = SEEK_END 13 if sz < 0 { return 0 } 14 return sz 15} 16// append a record to the primary log; returns the new end position. 17func rep_append(path: *u8, record: *u8, n: i64) -> i64 { 18 let fd: i64 = sys_openat_append(path, 0x1a4); if fd < 0 { return 0 - 1 } 19 sys_write(fd, record, n); sys_close(fd) 20 return rep_position(path) 21} 22func rep_read_all(path: *u8, out: *u8, cap: i64) -> i64 { 23 let fd: i64 = sys_openat_rd(path); if fd < 0 { return 0 } 24 var total: i64 = 0; var go: i64 = 1 25 while go == 1 { let nr: i64 = sys_read(fd, (out as i64 + total) as *u8, cap - total); if nr <= 0 { go = 0 } if nr > 0 { total = total + nr } if total >= cap { go = 0 } } 26 sys_close(fd); return total 27} 28// ship primary[from_pos .. end] and APPEND it to the standby. Returns bytes shipped. 29func rep_ship(primary: *u8, standby: *u8, from_pos: i64) -> i64 { 30 let fd: i64 = sys_openat_rd(primary); if fd < 0 { return 0 } 31 sys_lseek(fd, from_pos, 0) // SEEK_SET 32 let buf: *u8 = sys_mmap(K_MAGIC_1048576) 33 var total: i64 = 0; var go: i64 = 1 34 while go == 1 { let nr: i64 = sys_read(fd, (buf as i64 + total) as *u8, K_MAGIC_1048576 - total); if nr <= 0 { go = 0 } if nr > 0 { total = total + nr } if total >= K_MAGIC_1048576 { go = 0 } } 35 sys_close(fd) 36 if total > 0 { let sfd: i64 = sys_openat_append(standby, 0x1a4); if sfd >= 0 { sys_write(sfd, buf, total); sys_close(sfd) } } 37 return total 38} 39// replication lag in bytes (primary_end - standby_end); 0 = caught up. 40func rep_lag(primary: *u8, standby: *u8) -> i64 { return rep_position(primary) - rep_position(standby) } 41// 1 iff the standby is a BYTE-EXACT prefix of the primary (a valid replica), else 0. 42func rep_in_sync(primary: *u8, standby: *u8) -> i64 { 43 let pb: *u8 = sys_mmap(K_MAGIC_1048576); let pn: i64 = rep_read_all(primary, pb, K_MAGIC_1048576) 44 let sb: *u8 = sys_mmap(K_MAGIC_1048576); let sn: i64 = rep_read_all(standby, sb, K_MAGIC_1048576) 45 if sn > pn { return 0 } 46 var i: i64 = 0 47 while i < sn { if pb[i] != sb[i] { return 0 } i = i + 1 } 48 return 1 49}