"encoding/json"
"errors"
"fmt"
+ "log"
"os"
"path/filepath"
"strings"
"time"
)
-// errMemoryNoView tells that a chat has messages but no saved view.
-var errMemoryNoView = errors.New("the chat has messages but no saved view")
+var (
+ // errMemoryNoView tells that a chat has messages but no saved view.
+ errMemoryNoView = errors.New("the chat has messages but no saved view")
+
+ // errMemoryUnsettled tells that a line of the view stays pending,
+ // because its build failed.
+ errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message")
+)
// openMemory opens the chat in root and starts the builds that are ready. The
-// clock gives the dates of new messages and the days of new nodes. The builds
-// stop when ctx ends or at Close.
-func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer) (*Memory, error) {
+// clock gives the dates of new messages and the days of new nodes. logger gets
+// one line for each failed build. The builds stop when ctx ends or at Close.
+func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer, logger *log.Logger) (*Memory, error) {
log, err := openLog(ctx, root, clock)
if err != nil {
return nil, err
log: log,
queue: newBuildQueue(),
summarizer: summarizer,
+ logger: logger,
changed: make(chan struct{}),
}
cview *View
queue *BuildQueue
summarizer Summarizer
+ logger *log.Logger
ctx context.Context
cancel context.CancelFunc
workers sync.WaitGroup
}
}
+// Settle returns when the chat view has no pending line, so that every message
+// so far is summarized. A pending line with no build that runs belongs to a
+// failed build, which waits for the next message, so Settle returns
+// errMemoryUnsettled in place of a wait that never ends.
+func (m *Memory) Settle(ctx context.Context) error {
+ for {
+ m.mutex.Lock()
+ pending := m.chat.Pending()
+ running := m.queue.Running()
+ changed := m.changed
+ m.mutex.Unlock()
+
+ if !pending {
+ return nil
+ }
+
+ if running == 0 {
+ return errMemoryUnsettled
+ }
+
+ select {
+ case <-changed:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ }
+}
+
// Flush returns when every write before the call is durable.
func (m *Memory) Flush(ctx context.Context) error {
return m.syncer.Flush(ctx)
task, err := m.task(key)
if err != nil {
+ m.failed(key, fmt.Errorf("read its input: %w", err))
m.queue.Fail(key)
+
continue
}
m.syncer.Written()
}
+ if err != nil && m.ctx.Err() == nil {
+ m.failed(key, err)
+ }
+
if err != nil {
m.queue.Fail(key)
} else {
m.dispatch()
}
+// failed logs the failed build of the node key with its cause err.
+func (m *Memory) failed(key NodeKey, err error) {
+ m.logger.Printf("summary failed, retry at next message: node (%d, %d): %v", key.level, key.index, err)
+}
+
// ready tells if both children of the node key are built and the node is not.
func (m *Memory) ready(key NodeKey) (bool, error) {
for _, index := range []int{2 * key.index, 2*key.index + 1} {