amir/dreamstack
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: per-signal version counters for conflict resolution

Browse source 
- _signalVersions: monotonic counter per signal, incremented on each local mutation
- Diffs include _v: {name: version} for version comparison
- _applyRemoteDiff only applies if remote version >= local version
- Prevents stale overwrites when both devices edit simultaneously
- onopen snapshot includes version map for late-joiner consistency
This commit is contained in:
enzotar 2026-02-25 21:58:14 -08:00
parent 0290ed464a
commit e5ff612197
1 changed files with 17 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
compiler/ds-codegen/src/js_emitter.rs
View file

@ -1818,11 +1818,13 @@ const DS = (() => {
// ── Signal registry for bidirectional sync ── // ── Signal registry for bidirectional sync ──
var _signalRegistry = {}; var _signalRegistry = {};
var _signalVersions = {}; // per-signal version counters for conflict resolution
var _applyingRemoteDiff = false; var _applyingRemoteDiff = false;
var _peerId = Math.random().toString(36).substr(2, 8); // unique per client var _peerId = Math.random().toString(36).substr(2, 8); // unique per client
function _registerSignal(name, sig) { function _registerSignal(name, sig) {
_signalRegistry[name] = sig; _signalRegistry[name] = sig;
_signalVersions[name] = 0;
} }
function _applyRemoteDiff(json) { function _applyRemoteDiff(json) {
@ -1830,12 +1832,18 @@ const DS = (() => {
var data = JSON.parse(json); var data = JSON.parse(json);
// Ignore our own diffs echoed back by the relay // Ignore our own diffs echoed back by the relay
if (data._pid === _peerId) return; if (data._pid === _peerId) return;
var versions = data._v || {};
_applyingRemoteDiff = true; _applyingRemoteDiff = true;
for (var name in data) { for (var name in data) {
if (name === '_pid') continue; if (name === '_pid' || name === '_v') continue;
var sig = _signalRegistry[name]; var sig = _signalRegistry[name];
if (sig) { if (!sig) continue;
var remoteV = versions[name] || 0;
var localV = _signalVersions[name] || 0;
// Only apply if remote version >= local version (last-write-wins)
if (remoteV >= localV) {
sig.value = data[name]; sig.value = data[name];
_signalVersions[name] = remoteV;
} }
} }
flush(); flush();
@ -1869,11 +1877,12 @@ const DS = (() => {
_streamWs.onclose = function() { setTimeout(function() { _initStream(url, mode); }, 2000); }; _streamWs.onclose = function() { setTimeout(function() { _initStream(url, mode); }, 2000); };
_streamWs.onopen = function() { _streamWs.onopen = function() {
console.log('[ds-stream] Peer connected:', peerUrl); console.log('[ds-stream] Peer connected:', peerUrl);
// Broadcast full state snapshot so other peers can sync // Broadcast full state snapshot with versions so other peers can sync
var fullState = { _pid: _peerId }; var fullState = { _pid: _peerId, _v: {} };
for (var name in _signalRegistry) { for (var name in _signalRegistry) {
var sig = _signalRegistry[name]; var sig = _signalRegistry[name];
fullState[name] = (typeof sig === 'object' && sig !== null && '_value' in sig) ? sig._value : sig; fullState[name] = (typeof sig === 'object' && sig !== null && '_value' in sig) ? sig._value : sig;
fullState._v[name] = _signalVersions[name] || 0;
} }
_streamSend(0x31, 0, new TextEncoder().encode(JSON.stringify(fullState))); _streamSend(0x31, 0, new TextEncoder().encode(JSON.stringify(fullState)));
}; };
@ -1896,7 +1905,10 @@ const DS = (() => {
function _streamDiff(name, value) { function _streamDiff(name, value) {
if (!_streamWs || _streamWs.readyState !== 1 || _streamMode !== 'signal') return; if (!_streamWs || _streamWs.readyState !== 1 || _streamMode !== 'signal') return;
if (_applyingRemoteDiff) return; // prevent echo loops if (_applyingRemoteDiff) return; // prevent echo loops
var obj = { _pid: _peerId }; // Increment version for conflict resolution
_signalVersions[name] = (_signalVersions[name] || 0) + 1;
var obj = { _pid: _peerId, _v: {} };
obj._v[name] = _signalVersions[name];
obj[name] = (typeof value === 'object' && value !== null && 'value' in value) ? value.value : value; obj[name] = (typeof value === 'object' && value !== null && 'value' in value) ? value.value : value;
_streamSend(0x31, 0, new TextEncoder().encode(JSON.stringify(obj))); _streamSend(0x31, 0, new TextEncoder().encode(JSON.stringify(obj)));
} }