@@ -639,6 +639,43 @@ def __getattr__(self, name):
639639 return getattr (self .inner , name )
640640
641641
642+ class _RecordingStore :
643+ """Forwards to `inner`, recording the ref of every blob written through
644+ it — how `commit()` learns exactly which blobs a commit created, which
645+ a local-first backend's journal wants (`commit_root`). Trie copy-on-
646+ write means only changed paths and new values are written, so the
647+ recorded set is naturally the commit's new-blob list."""
648+
649+ def __init__ (self , inner : BytesStore ):
650+ self .inner = inner
651+ self .refs : set = set ()
652+
653+ def put (self , data : bytes ) -> Ref :
654+ ref = self .inner .put (data )
655+ self .refs .add (ref )
656+ return ref
657+
658+ def put_many (self , datas : Iterable [bytes ]) -> List [Ref ]:
659+ datas = list (datas )
660+ put_many = getattr (self .inner , "put_many" , None )
661+ refs = (put_many (datas ) if put_many
662+ else [self .inner .put (d ) for d in datas ])
663+ self .refs .update (refs )
664+ return refs
665+
666+ def get (self , ref : Ref ) -> bytes :
667+ return self .inner .get (ref )
668+
669+ def get_many (self , refs : Iterable [Ref ]) -> Dict [Ref , bytes ]:
670+ get_many = getattr (self .inner , "get_many" , None )
671+ if get_many is not None :
672+ return get_many (refs )
673+ return {r : self .inner .get (r ) for r in refs }
674+
675+ def __getattr__ (self , name ):
676+ return getattr (self .inner , name )
677+
678+
642679# ---------------------------------------------------------------------------
643680# Persistent compacted radix trie (canonical)
644681#
@@ -1747,6 +1784,24 @@ def _fetch_blobs(self, refs: List[Ref]) -> Dict[Ref, bytes]:
17471784
17481785 # -- commit ---------------------------------------------------------------
17491786
1787+ def _journal_commit (self , base : Optional [Ref ], new : Optional [Ref ],
1788+ vrec : "_RecordingStore" ,
1789+ nrec : "_RecordingStore" ) -> None :
1790+ """Record the commit in a local-first backend's journal (duck-typed:
1791+ the backend has `commit_root`, e.g. swarmfs's LocalStore). The
1792+ recorders captured exactly the blobs this commit wrote — values and
1793+ trie nodes separately, so nodes carry the `structure` eviction
1794+ hint. Skipped when the root didn't change or canonical addressing
1795+ brought the store back to an already-journaled state (an emptied
1796+ store — root None — cannot be journaled either: the format has no
1797+ null-root event; keep a pointer for the head in that case)."""
1798+ inner = self ._blobs
1799+ if new is None or new == base or inner .has_root (new ):
1800+ return
1801+ parent = base if (base is None or inner .has_root (base )) else None
1802+ inner .commit_root (new , parent , sorted (vrec .refs | nrec .refs ),
1803+ structure = sorted (nrec .refs - vrec .refs ))
1804+
17501805 def _build_root (self , base : Optional [Ref ]) -> Optional [Ref ]:
17511806 """Apply the staged changes on top of `base` and return the new root.
17521807 Value blobs go up front (concurrently if supported); trie nodes are
@@ -1792,12 +1847,28 @@ def commit(self, *, reconcile: bool = False, resolver=None,
17921847 if self ._readonly :
17931848 raise TypeError ("read-only snapshot" )
17941849 base = self ._root
1795- new = self ._build_root (base )
1796- if self ._pointer is not None :
1797- if reconcile :
1798- new = self ._reconcile (base , new , resolver , retries )
1799- else :
1800- self ._pointer .set (new )
1850+ inner = self ._blobs
1851+ journaled = hasattr (inner , "commit_root" ) # local-first backend?
1852+ vrec = nrec = None
1853+ if journaled :
1854+ # Record which blobs this commit writes — values through
1855+ # self._blobs, trie nodes through the trie's handle — so the
1856+ # journal event can list them (and classify the nodes as
1857+ # `structure`). The recorders forward everything else.
1858+ vrec , nrec = _RecordingStore (inner ), _RecordingStore (inner )
1859+ self ._blobs , self ._trie ._blobs = vrec , nrec
1860+ try :
1861+ new = self ._build_root (base )
1862+ if self ._pointer is not None :
1863+ if reconcile :
1864+ new = self ._reconcile (base , new , resolver , retries )
1865+ else :
1866+ self ._pointer .set (new )
1867+ finally :
1868+ if journaled :
1869+ self ._blobs , self ._trie ._blobs = inner , inner
1870+ if journaled :
1871+ self ._journal_commit (base , new , vrec , nrec )
18011872 self ._staged .clear ()
18021873 self ._root = new
18031874 return new
@@ -1901,3 +1972,115 @@ def merge(cls, bytes_store: BytesStore, base: Optional[Ref],
19011972 finally :
19021973 trie ._reset_buffer ()
19031974 return root
1975+
1976+
1977+ # ---------------------------------------------------------------------------
1978+ # Local-first: RecordStore over a swarmfs localstore directory
1979+ # ---------------------------------------------------------------------------
1980+
1981+ class LocalFirstRecordStore (RecordStore ):
1982+ """A RecordStore whose backend is a local-first store directory —
1983+ create it with :func:`local_first_store`.
1984+
1985+ Commits land on local disk instantly (offline is the normal mode) and
1986+ are recorded in the store directory's journal — the reflog: lineage,
1987+ durability rungs, everything `sync_status()` reports. When opened with
1988+ an ``api_url``, a background syncer pushes commits to Swarm and
1989+ confirms them peer-to-peer; ``sync()`` is the blocking certainty
1990+ barrier ("my data is really out there"). Reads of locally evicted
1991+ blobs heal transparently by verified re-fetch.
1992+ """
1993+
1994+ def __init__ (self , bytes_store , local , syncer = None , ** kw ):
1995+ super ().__init__ (bytes_store , ** kw )
1996+ #: The underlying swarmfs LocalStore (journal, budget, pins).
1997+ self .local = local
1998+ #: The background pusher, or None when opened without api_url.
1999+ self .syncer = syncer
2000+
2001+ def sync (self , timeout : Optional [float ] = None ) -> None :
2002+ """Block until every commit is network-confirmed — the fsync of
2003+ the durability ladder. TimeoutError (naming the last sync error)
2004+ if `timeout` passes first."""
2005+ if self .syncer is None :
2006+ raise RuntimeError (
2007+ "this store was opened without api_url — local-only; "
2008+ "reopen with api_url=... to sync to Swarm" )
2009+ self .syncer .sync (timeout )
2010+
2011+ def sync_status (self ):
2012+ """The local-first store's status: bytes pinned (unpushed) vs
2013+ evictable, each root's durability rung, blobs living only on
2014+ Swarm. See swarmfs's ``StoreStatus``."""
2015+ return self .local .status ()
2016+
2017+ def close (self ) -> None :
2018+ if self .syncer is not None :
2019+ self .syncer .stop ()
2020+ self .local .close ()
2021+
2022+ def __enter__ (self ) -> "LocalFirstRecordStore" :
2023+ return self
2024+
2025+ def __exit__ (self , * exc ) -> None :
2026+ self .close ()
2027+
2028+
2029+ def local_first_store (path : str , api_url : Optional [str ] = None , * ,
2030+ stamp : str = "auto" ,
2031+ max_bytes : Optional [int ] = None ,
2032+ cache_bytes : int = 64 * 1024 * 1024 ,
2033+ addressing : str = "swarm" ,
2034+ sync_policy = None , witness = None ,
2035+ node_cache_size : int = DEFAULT_NODE_CACHE_SIZE
2036+ ) -> LocalFirstRecordStore :
2037+ """Open (or create) a local-first record store: commits go to local
2038+ disk instantly, a background worker pushes them to Swarm and confirms
2039+ arrival, and local storage behaves as a budgeted working set.
2040+
2041+ The one call for the whole arrangement::
2042+
2043+ store = local_first_store("~/.myapp/store", "http://localhost:1633")
2044+ store.put("k", {"v": 1})
2045+ store.commit() # local, instant, offline-safe
2046+ store.sync() # optional barrier: confirmed on Swarm
2047+
2048+ Without ``api_url`` the store is local-only (commits journal as usual
2049+ and push later, when reopened with an ``api_url``). ``max_bytes``
2050+ budgets the directory — unpushed data is pinned and the limit is soft
2051+ for it; only Swarm-confirmed blobs are evicted under pressure, and
2052+ reads of evicted blobs heal by verified re-fetch. ``cache_bytes``
2053+ sizes the in-memory blob cache (0 disables). ``stamp``/``sync_policy``
2054+ /``witness`` pass through to swarmfs's ``BeeRemote``/``Syncer``.
2055+
2056+ Head resolution: the journal records lineage; a ``HEAD`` pointer file
2057+ in the directory tracks the current root (needed because canonical
2058+ addressing means returning to a previous state re-uses its old root,
2059+ which the append-only journal deliberately refuses to duplicate).
2060+
2061+ Requires swarmfs >= 0.5 (``pip install "recordstore[local]"``).
2062+ Feed publication is not wired here yet — publishing a head to a Swarm
2063+ feed belongs *after* confirmation, and lands with the R2 phase.
2064+ """
2065+ try :
2066+ from swarmfs .localstore import LocalStore
2067+ except ImportError as e :
2068+ raise ImportError (
2069+ "local_first_store needs swarmfs with its localstore module "
2070+ '(>= 0.5): pip install "recordstore[local]"' ) from e
2071+ if api_url is not None and addressing != "swarm" :
2072+ raise ValueError (
2073+ 'pushing to Swarm requires addressing="swarm" (the push '
2074+ "asserts the node returns the locally computed reference)" )
2075+ local = LocalStore (path , addressing = addressing , max_bytes = max_bytes )
2076+ syncer = None
2077+ if api_url is not None :
2078+ from swarmfs .localsync import BeeRemote , Syncer
2079+ remote = BeeRemote (api_url , stamp = stamp )
2080+ syncer = Syncer (local , remote , sync_policy , witness = witness ).start ()
2081+ blobs = CachedBytesStore (local , cache_bytes ) if cache_bytes else local
2082+ pointer = FilePointer (os .path .join (local .path , "HEAD" ))
2083+ root = pointer .get () if pointer .get () is not None else local .latest_root ()
2084+ return LocalFirstRecordStore (blobs , local , syncer , root = root ,
2085+ pointer = pointer ,
2086+ node_cache_size = node_cache_size )
0 commit comments