# dirstate.py - working directory tracking for mercurial

#

# Copyright 2005-2007 Olivia Mackall <olivia@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

import collections

import contextlib

import os

import stat

import uuid

from .i18n import _

from .pycompat import delattr

from hgdemandimport import tracing

from . import (

    dirstatemap,

    encoding,

    error,

    match as matchmod,

    node,

    pathutil,

    policy,

    pycompat,

    scmutil,

    util,

)

from .dirstateutils import (

    timestamp,

)

from .interfaces import (

    dirstate as intdirstate,

    util as interfaceutil,

)

parsers = policy.importmod('parsers')

rustmod = policy.importrust('dirstate')

HAS_FAST_DIRSTATE_V2 = rustmod is not None

propertycache = util.propertycache

filecache = scmutil.filecache

_rangemask = dirstatemap.rangemask

DirstateItem = dirstatemap.DirstateItem

class repocache(filecache):

    """filecache for files in .hg/"""

    def join(self, obj, fname):

        return obj._opener.join(fname)

class rootcache(filecache):

    """filecache for files in the repository root"""

    def join(self, obj, fname):

        return obj._join(fname)

def check_invalidated(func):

    """check that the func is called with a non-invalidated dirstate

    The dirstate is in an "invalidated state" after an error occured during its

    modification and remains so until we exited the top level scope that framed

    such change.

    """

    def wrap(self, *args, **kwargs):

        if self._invalidated_context:

            msg = 'calling `%s` after the dirstate was invalidated'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return wrap

def requires_changing_parents(func):

    def wrap(self, *args, **kwargs):

        if not self.is_changing_parents:

            msg = 'calling `%s` outside of a changing_parents context'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return check_invalidated(wrap)

def requires_changing_files(func):

    def wrap(self, *args, **kwargs):

        if not self.is_changing_files:

            msg = 'calling `%s` outside of a `changing_files`'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return check_invalidated(wrap)

def requires_changing_any(func):

    def wrap(self, *args, **kwargs):

        if not self.is_changing_any:

            msg = 'calling `%s` outside of a changing context'

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return check_invalidated(wrap)

def requires_changing_files_or_status(func):

    def wrap(self, *args, **kwargs):

        if not (self.is_changing_files or self._running_status > 0):

            msg = (

                'calling `%s` outside of a changing_files '

                'or running_status context'

            )

            msg %= func.__name__

            raise error.ProgrammingError(msg)

        return func(self, *args, **kwargs)

    return check_invalidated(wrap)

CHANGE_TYPE_PARENTS = "parents"

CHANGE_TYPE_FILES = "files"

@interfaceutil.implementer(intdirstate.idirstate)

class dirstate:

    # used by largefile to avoid overwritting transaction callback

    _tr_key_suffix = b''

    def __init__(

        self,

        opener,

        ui,

        root,

        validate,

        sparsematchfn,

        nodeconstants,

        use_dirstate_v2,

        use_tracked_hint=False,

    ):

        """Create a new dirstate object.

        opener is an open()-like callable that can be used to open the

        dirstate file; root is the root of the directory tracked by

        the dirstate.

        """

        self._use_dirstate_v2 = use_dirstate_v2

        self._use_tracked_hint = use_tracked_hint

        self._nodeconstants = nodeconstants

        self._opener = opener

        self._validate = validate

        self._root = root

        # Either build a sparse-matcher or None if sparse is disabled

        self._sparsematchfn = sparsematchfn

        # ntpath.join(root, '') of Python 2.7.9 does not add sep if root is

        # UNC path pointing to root share (issue4557)

        self._rootdir = pathutil.normasprefix(root)

        # True is any internal state may be different

        self._dirty = False

        # True if the set of tracked file may be different

        self._dirty_tracked_set = False

        self._ui = ui

        self._filecache = {}

        # nesting level of `changing_parents` context

        self._changing_level = 0

        # the change currently underway

        self._change_type = None

        # number of open _running_status context

        self._running_status = 0

        # True if the current dirstate changing operations have been

        # invalidated (used to make sure all nested contexts have been exited)

        self._invalidated_context = False

        self._attached_to_a_transaction = False

        self._filename = b'dirstate'

        self._filename_th = b'dirstate-tracked-hint'

        self._pendingfilename = b'%s.pending' % self._filename

        self._plchangecallbacks = {}

        self._origpl = None

        self._mapcls = dirstatemap.dirstatemap

        # Access and cache cwd early, so we don't access it for the first time

        # after a working-copy update caused it to not exist (accessing it then

        # raises an exception).

        self._cwd

    def refresh(self):

        if '_branch' in vars(self):

            del self._branch

        if '_map' in vars(self) and self._map.may_need_refresh():

            self.invalidate()

    def prefetch_parents(self):

        """make sure the parents are loaded

        Used to avoid a race condition.

        """

        self._pl

    @contextlib.contextmanager

    @check_invalidated

    def running_status(self, repo):

        """Wrap a status operation

        This context is not mutally exclusive with the `changing_*` context. It

        also do not warrant for the `wlock` to be taken.

        If the wlock is taken, this context will behave in a simple way, and

        ensure the data are scheduled for write when leaving the top level

        context.

        If the lock is not taken, it will only warrant that the data are either

        committed (written) and rolled back (invalidated) when exiting the top

        level context. The write/invalidate action must be performed by the

        wrapped code.

        The expected  logic is:

        A: read the dirstate

        B: run status

           This might make the dirstate dirty by updating cache,

           especially in Rust.

        C: do more "post status fixup if relevant

        D: try to take the w-lock (this will invalidate the changes if they were raced)

        E0: if dirstate changed on disk → discard change (done by dirstate internal)

        E1: elif lock was acquired → write the changes

        E2: else → discard the changes

        """

        has_lock = repo.currentwlock() is not None

        is_changing = self.is_changing_any

        tr = repo.currenttransaction()

        has_tr = tr is not None

        nested = bool(self._running_status)

        first_and_alone = not (is_changing or has_tr or nested)

        # enforce no change happened outside of a proper context.

        if first_and_alone and self._dirty:

            has_tr = repo.currenttransaction() is not None

            if not has_tr and self._changing_level == 0 and self._dirty:

                msg = "entering a status context, but dirstate is already dirty"

                raise error.ProgrammingError(msg)

        should_write = has_lock and not (nested or is_changing)

        self._running_status += 1

        try:

            yield

        except Exception:

            self.invalidate()

            raise

        finally:

            self._running_status -= 1

            if self._invalidated_context:

                should_write = False

                self.invalidate()

        if should_write:

            assert repo.currenttransaction() is tr

            self.write(tr)

        elif not has_lock:

            if self._dirty:

                msg = b'dirstate dirty while exiting an isolated status context'

                repo.ui.develwarn(msg)

                self.invalidate()

    @contextlib.contextmanager

    @check_invalidated

    def _changing(self, repo, change_type):

        if repo.currentwlock() is None:

            msg = b"trying to change the dirstate without holding the wlock"

            raise error.ProgrammingError(msg)

        has_tr = repo.currenttransaction() is not None

        if not has_tr and self._changing_level == 0 and self._dirty:

            msg = b"entering a changing context, but dirstate is already dirty"

            repo.ui.develwarn(msg)

        assert self._changing_level >= 0

        # different type of change are mutually exclusive

        if self._change_type is None:

            assert self._changing_level == 0

            self._change_type = change_type

        elif self._change_type != change_type:

            msg = (

                'trying to open "%s" dirstate-changing context while a "%s" is'

                ' already open'

            )

            msg %= (change_type, self._change_type)

            raise error.ProgrammingError(msg)

        should_write = False

        self._changing_level += 1

        try:

            yield

        except:  # re-raises

            self.invalidate()  # this will set `_invalidated_context`

            raise

        finally:

            assert self._changing_level > 0

            self._changing_level -= 1

            # If the dirstate is being invalidated, call invalidate again.

            # This will throw away anything added by a upper context and

            # reset the `_invalidated_context` flag when relevant

            if self._changing_level <= 0:

                self._change_type = None

                assert self._changing_level == 0

            if self._invalidated_context:

                # make sure we invalidate anything an upper context might

                # have changed.

                self.invalidate()

            else:

                should_write = self._changing_level <= 0

        tr = repo.currenttransaction()

        if has_tr != (tr is not None):

            if has_tr:

                m = "transaction vanished while changing dirstate"

            else:

                m = "transaction appeared while changing dirstate"

            raise error.ProgrammingError(m)

        if should_write:

            self.write(tr)

    @contextlib.contextmanager

    def changing_parents(self, repo):

        with self._changing(repo, CHANGE_TYPE_PARENTS) as c:

            yield c

    @contextlib.contextmanager

    def changing_files(self, repo):

        with self._changing(repo, CHANGE_TYPE_FILES) as c:

            yield c

    # here to help migration to the new code

    def parentchange(self):

        msg = (

            "Mercurial 6.4 and later requires call to "

            "`dirstate.changing_parents(repo)`"

        )

        raise error.ProgrammingError(msg)

    @property

    def is_changing_any(self):

        """Returns true if the dirstate is in the middle of a set of changes.

        This returns True for any kind of change.

        """

        return self._changing_level > 0

    def pendingparentchange(self):

        return self.is_changing_parent()

    def is_changing_parent(self):

        """Returns true if the dirstate is in the middle of a set of changes

        that modify the dirstate parent.

        """

        self._ui.deprecwarn(b"dirstate.is_changing_parents", b"6.5")

        return self.is_changing_parents

    @property

    def is_changing_parents(self):

        """Returns true if the dirstate is in the middle of a set of changes

        that modify the dirstate parent.

        """

        if self._changing_level <= 0:

            return False

        return self._change_type == CHANGE_TYPE_PARENTS

    @property

    def is_changing_files(self):

        """Returns true if the dirstate is in the middle of a set of changes

        that modify the files tracked or their sources.

        """

        if self._changing_level <= 0:

            return False

        return self._change_type == CHANGE_TYPE_FILES

    @propertycache

    def _map(self):

        """Return the dirstate contents (see documentation for dirstatemap)."""

        return self._mapcls(

            self._ui,

            self._opener,

            self._root,

            self._nodeconstants,

            self._use_dirstate_v2,

        )

    @property