Bind external events to permanent agent IDs

[onechat] / syncer.go

diff --git a/syncer.go b/syncer.go
index 3da6d7a6dec0daa7ad2d6a9ae4e004445f694e91..297405bd6d0eccdcfd9bb771050b3ff67eb7e0cd 100644 (file)
--- a/syncer.go
+++ b/syncer.go
@@ -10,22 +10,6 @@ import (
        "golang.org/x/sys/unix"
 )
 
        "golang.org/x/sys/unix"
 )
 

-// openSyncer returns the syncer of the file system that holds dir. It syncs
-// with syncfs, which writes back only that file system.
-func openSyncer(dir string) (*Syncer, error) {
-       file, err := os.Open(dir)
-       if err != nil {
-               return nil, fmt.Errorf("open chat directory: %w", err)
-       }
-
-       s := newSyncer(func() error {
-               return unix.Syncfs(int(file.Fd()))
-       })
-       s.file = file
-
-       return s, nil
-}
-

 // Syncer makes the writes of a chat durable by group commit. Each write goes to
 // the operating system at once, so a crash of the process loses nothing. After
 // each write, Written starts a background sync of the file system of the chat,
 // Syncer makes the writes of a chat durable by group commit. Each write goes to
 // the operating system at once, so a crash of the process loses nothing. After
 // each write, Written starts a background sync of the file system of the chat,


@@ -48,59 +32,39 @@ type Syncer struct {
        workers sync.WaitGroup
 }
 
        workers sync.WaitGroup
 }
 

-// Written tells the syncer that a write went to the operating system, and
-// starts a sync when none runs.
-func (s *Syncer) Written() {
-       s.mutex.Lock()
-       defer s.mutex.Unlock()
-
-       s.written++
-       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()
+// openSyncer returns the syncer of the file system that holds dir. It syncs
+// with syncfs, which writes back only that file system.
+func openSyncer(dir string) (*Syncer, error) {
+       file, err := os.Open(dir)
+       if err != nil {
+               return nil, fmt.Errorf("open chat directory: %w", err)
+       }

 
 

-               reached := s.durable >= target
-               failure := s.failure
-               done := s.done
-               s.mutex.Unlock()
+       s := newSyncer(func() error {
+               return unix.Syncfs(int(file.Fd()))
+       })

 
 

-               if reached {
-                       return nil
-               }
+       s.file = file

 
 

-               if failure != nil {
-                       return failure
-               }
+       return s, nil
+}

 
 

-               select {
-               case <-done:
-               case <-ctx.Done():
-                       return ctx.Err()
-               }
+// newSyncer returns a syncer that makes the writes durable with sync.
+func newSyncer(sync func() error) *Syncer {
+       return &Syncer{
+               sync: sync,
+               done: make(chan struct{}),

        }
 }
 
        }
 }
 

-// 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
-       }
+// Written tells the syncer that a write went to the operating system, and
+// starts a sync when none runs.
+func (s *Syncer) Written() {
+       s.mutex.Lock()
+       defer s.mutex.Unlock()

 
 

-       return err
+       s.written++
+       s.start()

 }
 
 // start starts a sync of the writes so far when none runs and a write is not
 }
 
 // start starts a sync of the writes so far when none runs and a write is not


@@ -139,10 +103,47 @@ func (s *Syncer) run(target uint64) {
        s.start()
 }
 
        s.start()
 }
 

-// newSyncer returns a syncer that makes the writes durable with sync.
-func newSyncer(sync func() error) *Syncer {
-       return &Syncer{
-               sync: sync,
-               done: make(chan struct{}),
+// 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()
+               }

        }
 }
        }
 }