Bind external events to permanent agent IDs

[onechat] / syncer.go

diff --git a/syncer.go b/syncer.go
index 0784393fcd4df9bbfc44e8d99877eaca9f3bac71..297405bd6d0eccdcfd9bb771050b3ff67eb7e0cd 100644 (file)
--- a/syncer.go
+++ b/syncer.go
@@ -43,6 +43,7 @@ func openSyncer(dir string) (*Syncer, error) {
        s := newSyncer(func() error {
                return unix.Syncfs(int(file.Fd()))
        })
        s := newSyncer(func() error {
                return unix.Syncfs(int(file.Fd()))
        })

+

        s.file = file
 
        return s, nil
        s.file = file
 
        return s, nil


@@ -66,51 +67,6 @@ func (s *Syncer) Written() {
        s.start()
 }
 
        s.start()
 }
 

-// Flush returns when every write before the call is durable.
-func (s *Syncer) Flush(ctx context.Context) error {
-       s.mutex.Lock()
-
-       target := s.written
-       s.mutex.Unlock()
-
-       for {
-               s.mutex.Lock()
-
-               reached := s.durable >= target
-               failure := s.failure
-               done := s.done
-               s.mutex.Unlock()
-
-               if reached {
-                       return nil
-               }
-
-               if failure != nil {
-                       return failure
-               }
-
-               select {
-               case <-done:
-               case <-ctx.Done():
-                       return ctx.Err()
-               }
-       }
-}
-
-// Close makes every write durable, waits for the syncs, and closes the
-// directory.
-func (s *Syncer) Close() error {
-       err := s.Flush(context.Background())
-       s.workers.Wait()
-
-       if s.file != nil {
-               err = errors.Join(err, s.file.Close())
-               s.file = nil
-       }
-
-       return err
-}
-

 // start starts a sync of the writes so far when none runs and a write is not
 // durable. The caller holds the mutex.
 func (s *Syncer) start() {
 // start starts a sync of the writes so far when none runs and a write is not
 // durable. The caller holds the mutex.
 func (s *Syncer) start() {


@@ -146,3 +102,48 @@ func (s *Syncer) run(target uint64) {
        s.done = make(chan struct{})
        s.start()
 }
        s.done = make(chan struct{})
        s.start()
 }

+
+// Close makes every write durable, waits for the syncs, and closes the
+// directory.
+func (s *Syncer) Close() error {
+       err := s.Flush(context.Background())
+       s.workers.Wait()
+
+       if s.file != nil {
+               err = errors.Join(err, s.file.Close())
+               s.file = nil
+       }
+
+       return err
+}
+
+// Flush returns when every write before the call is durable.
+func (s *Syncer) Flush(ctx context.Context) error {
+       s.mutex.Lock()
+
+       target := s.written
+       s.mutex.Unlock()
+
+       for {
+               s.mutex.Lock()
+
+               reached := s.durable >= target
+               failure := s.failure
+               done := s.done
+               s.mutex.Unlock()
+
+               if reached {
+                       return nil
+               }
+
+               if failure != nil {
+                       return fmt.Errorf("flush durable writes: %w", failure)
+               }
+
+               select {
+               case <-done:
+               case <-ctx.Done():
+                       return ctx.Err()
+               }
+       }
+}