Commit b6bbc50d authored by Kirill Smelkov's avatar Kirill Smelkov

.

parent 05be8f76
...@@ -71,21 +71,20 @@ func (e *baseMutatedError) Error() string { ...@@ -71,21 +71,20 @@ func (e *baseMutatedError) Error() string {
return fmt.Sprintf("base.head mutated from @%s to @%s", e.baseAt0, e.baseHead) return fmt.Sprintf("base.head mutated from @%s to @%s", e.baseAt0, e.baseHead)
} }
// baseError is reported on error detected on Storage.base . // baseErrorEvent is reported on error event detected on Storage.base .
type baseError struct { type baseErrorEvent struct {
err error err error
} }
func (e *baseError) Error() string { func (e *baseErrorEvent) Error() string {
return fmt.Sprintf("base: error: %s", e.err) return fmt.Sprintf("base: error event: %s", e.err)
} }
func (e *baseError) Cause() error { return e.err } func (e *baseErrorEvent) Cause() error { return e.err }
func (e *baseError) Unwrap() error { return e.err } func (e *baseErrorEvent) Unwrap() error { return e.err }
// watcher detects base mutation and proxies δ events to user watchq. // watcher task detects base mutation and proxies δ events to user watchq.
// it runs as separate goroutine.
func (d *Storage) watcher(ctx context.Context) error { func (d *Storage) watcher(ctx context.Context) error {
if d.watchq != nil { if d.watchq != nil {
defer close(d.watchq) defer close(d.watchq)
...@@ -97,7 +96,7 @@ func (d *Storage) watcher(ctx context.Context) error { ...@@ -97,7 +96,7 @@ func (d *Storage) watcher(ctx context.Context) error {
case <-ctx.Done(): case <-ctx.Done():
return ctx.Err() return ctx.Err()
// event on base -> base mutated + shutdown // event on base -> base mutated/error + shutdown
case event, ok := <-d.baseWatchq: case event, ok := <-d.baseWatchq:
if !ok { if !ok {
// base closed // base closed
...@@ -111,7 +110,7 @@ func (d *Storage) watcher(ctx context.Context) error { ...@@ -111,7 +110,7 @@ func (d *Storage) watcher(ctx context.Context) error {
edown = &baseMutatedError{d.baseAt0, event.Tid} edown = &baseMutatedError{d.baseAt0, event.Tid}
case *zodb.EventError: case *zodb.EventError:
edown = &baseError{event.Err} edown = &baseErrorEvent{event.Err}
default: default:
edown = fmt.Errorf("base: unexpected event %T", event) edown = fmt.Errorf("base: unexpected event %T", event)
...@@ -138,7 +137,7 @@ func (d *Storage) watcher(ctx context.Context) error { ...@@ -138,7 +137,7 @@ func (d *Storage) watcher(ctx context.Context) error {
} }
} }
// shutdown marks Storage as no longer being operational due to reason. // shutdown marks Storage as no longer operational due to reason.
func (d *Storage) shutdown(reason error) { func (d *Storage) shutdown(reason error) {
d.downOnce.Do(func() { d.downOnce.Do(func() {
d.downErr = reason d.downErr = reason
...@@ -160,7 +159,7 @@ func (d *Storage) Close() (err error) { ...@@ -160,7 +159,7 @@ func (d *Storage) Close() (err error) {
errδ := d.δ.Close() errδ := d.δ.Close()
errBase := d.base.Close() errBase := d.base.Close()
// cancel watcher. don't propagate its error to Close - watcher error // cancel watcher; don't propagate its error to Close - watcher error
// goes to watchq and op errors. // goes to watchq and op errors.
d.watchCancel() d.watchCancel()
_ = d.watchWG.Wait() _ = d.watchWG.Wait()
...@@ -254,8 +253,8 @@ func (d *Storage) Load(ctx context.Context, xid zodb.Xid) (_ *mem.Buf, _ zodb.Ti ...@@ -254,8 +253,8 @@ func (d *Storage) Load(ctx context.Context, xid zodb.Xid) (_ *mem.Buf, _ zodb.Ti
} }
} }
// cap .at in xid to .baseAt0 (we convert it back on error return, and // cap xid.At to .baseAt0 (we convert it back on error return, and
// it makes more robust wrt simultaneous base mutation). // it makes Load more robust wrt simultaneous base mutation).
xidBase := xid xidBase := xid
if xid.At > d.baseAt0 { if xid.At > d.baseAt0 {
xidBase.At = d.baseAt0 xidBase.At = d.baseAt0
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment